Tornado 阻塞异步请求

Posted 2023-02-24

【中文标题】Tornado 阻塞异步请求

【英文标题】：Tornado blocking asynchronous requests

【发布时间】：2012-10-14 15:12:02

【问题描述】：

使用 Tornado，我有一个 Get 请求需要很长时间，因为它向另一个 Web 服务发出许多请求并处理数据，可能需要几分钟才能完全完成。我不希望这会阻止整个 Web 服务器响应它当前所做的其他请求。

据我了解，Tornado 是单线程的，并且同步执行每个请求，即使它异步处理它们（仍然对此感到困惑）。漫长过程的某些部分可能是暂停点，以允许服务器处理其他请求（可能的解决方案？）。我在 Heroku 上用一个工人运行它，所以不确定这如何转化为产生一个新线程或多处理，我没有使用 python 的经验。

这是我正在尝试做的事情：客户端发出 get 调用以启动进程，然后我每 5 秒循环一次 get 调用以检查状态并使用新信息更新页面（长轮询也会工作但遇到同样的问题）。问题是启动长进程会阻塞所有新的获取请求（或新的长轮询会话），直到它完成。

有没有一种简单的方法可以启动这个漫长的 get 调用，并且不会让它在此过程中阻塞整个 Web 服务器？有什么我可以在代码中说的......“暂停，处理待处理的请求然后继续”？

我需要在 ProcessHandler 上发起一个获取请求。然后我需要在 ProcessHandler 运行时继续能够查询 StatusHandler。

例子：

class StatusHandler(tornado.web.RequestHandler):
    @tornado.web.asynchronous
    def get(self):
       self.render("status.html")

class ProcessHandler(tornado.web.RequestHandler):
    @tornado.web.asynchronous
    def get(self):
       self.updateStatus("0")
       result1 = self.function1()
       self.updateStatus("1")
       result2 = self.function2(result1)
       self.updateStatus("2")
       result3 = self.function3(result2)
       self.updateStatus("3")
       self.finish()

【问题讨论】：

你试过tornado.gen模块吗？ tornadoweb.org/documentation/gen.html

您是否记得将其注释为异步调用：在您的 GET 方法上添加：@asynchronous

andyboot 是的，我的 GET 方法上有 @asynchronous

唐，我试图将函数包装在 get.task 中，但它仍然阻止了其他 get 请求。我更新了我的帖子，以便更好地了解我正在尝试做的事情。

【参考方案1】：

这是一个完整的 Tornado 应用示例，它使用异步 HTTP 客户端和 gen.Task 模块使事情变得简单。

如果您在文档中阅读更多关于 gen.Task 的信息，您会发现实际上您可以同时调度多个请求。这是使用 Tornado 的核心思想，即一切都没有阻塞并且仍然保持单个进程。

更新：我添加了一个线程处理程序来演示如何将工作分派到第二个线程并在完成后接收callback()。

import os
import threading
import tornado.options
import tornado.ioloop
import tornado.httpserver
import tornado.httpclient
import tornado.web
from tornado import gen
from tornado.web import asynchronous

tornado.options.define('port', type=int, default=9000, help='server port number (default: 9000)')
tornado.options.define('debug', type=bool, default=False, help='run in debug mode with autoreload (default: False)')

class Worker(threading.Thread):
   def __init__(self, callback=None, *args, **kwargs):
        super(Worker, self).__init__(*args, **kwargs)
        self.callback = callback

   def run(self):
        import time
        time.sleep(10)
        self.callback('DONE')

class Application(tornado.web.Application):
    def __init__(self):
        handlers = [
            (r"/", IndexHandler),
            (r"/thread", ThreadHandler),
        ]
        settings = dict(
            static_path = os.path.join(os.path.dirname(__file__), "static"),
            template_path = os.path.join(os.path.dirname(__file__), "templates"),
            debug = tornado.options.options.debug,
        )
        tornado.web.Application.__init__(self, handlers, **settings)

class IndexHandler(tornado.web.RequestHandler):
    client = tornado.httpclient.AsyncHTTPClient()

    @asynchronous
    @gen.engine
    def get(self):
        response = yield gen.Task(self.client.fetch, "http://google.com")

        self.finish("Google's homepage is %d bytes long" % len(response.body))

class ThreadHandler(tornado.web.RequestHandler):
    @asynchronous
    def get(self):
        Worker(self.worker_done).start()

    def worker_done(self, value):
        self.finish(value)

def main():
    tornado.options.parse_command_line()
    http_server = tornado.httpserver.HTTPServer(Application())
    http_server.listen(tornado.options.options.port)
    tornado.ioloop.IOLoop.instance().start()

if __name__ == "__main__":
    main()

【讨论】：

我将函数包装在 gen.Task 中，但它仍然做同样的事情。我创建了一个有多个响应 = get.Tasks() 的 get。我不需要同时执行它们。实际上它们需要是串行的，但是在此获取请求正在进行时，任何其他获取请求都会被阻止。

我已经更新了我上面的例子。我试图用 gen.Task() 包装所有函数并且一切正常，但它仍然阻止我在完成之前响应 StatusHandler 上的查询。

在您的示例中 self.function1() 是一个 pure python 函数，它不会对外部服务进行其他调用？最初的假设是它调用了另一个服务，而你的服务被阻止了。

就是这样 - 您需要将 urllib2.urlopen() 函数替换为 Tornado AsyncHTTPClient 版本。由于 urlopen() 会阻塞直到接收到数据，而 AsyncHTTPClient 会将控制权返回给 ioloop。

感谢您的解决方案。但是，如果 /thread 正在加载，并且您为 /thread 打开另一个选项卡，它仍然会阻塞。为什么？

【参考方案2】：

koblas 的解决方案很棒。这是使用tornado.gen

的替代方法

import tornado.ioloop
import tornado.web
import tornado.gen
import tornado.concurrent
import time
from threading import Thread
from functools import wraps

def run_async(func):
  @wraps(func)
  def async_func(*args, **kwargs):
    func_hl = Thread(target = func, args = args, kwargs = kwargs)
    func_hl.start()
    return func_hl

  return async_func

@run_async
def sleeper(callback):
  i = 0
  while i <= 10:
    print i
    time.sleep(1)
    i += 1
  callback('DONE')


class MainHandler(tornado.web.RequestHandler):
    @tornado.web.asynchronous
    @tornado.gen.coroutine
    def get(self):
        response = yield tornado.gen.Task(sleeper)
        self.write(response)
        self.finish()

class OtherHandler(tornado.web.RequestHandler):
    def get(self):
        self.write('hello world')
        print 'in other'
        self.finish()