如何在同步环境中执行 Tornado 协程？答案

【问题标题】：How to execute Tornado coroutine inside of synchronous environment?如何在同步环境中执行 Tornado 协程？
【发布时间】：2016-03-29 14:16:33
【问题描述】：

我有一些 Tornado 的协程相关问题。

有一些python-model A，有能力执行一些功能。该功能可以从模型外部设置。我不能改变模型本身，但我可以传递任何我想要的函数。我正在尝试通过我的函数教它与 Tornado 的 ioloop 一起工作，但我做不到。

这里是sn-p：

import functools
import pprint
from tornado import gen
from tornado import ioloop

class A:
    f = None
    def execute(self):
      return self.f()
    pass

@gen.coroutine
def genlist():
    raise gen.Return(range(1, 10))

@gen.coroutine
def some_work():
    a = A()
    a.f = functools.partial(
        ioloop.IOLoop.instance().run_sync,
        lambda: genlist())
    print "a.f set"
    raise gen.Return(a)

@gen.coroutine
def main():
    a = yield some_work()
    retval = a.execute()
    raise gen.Return(retval)


if __name__ == "__main__":
    pprint.pprint(ioloop.IOLoop.current().run_sync(main))

所以问题是我在一部分代码中设置了函数，但在另一部分用模型的方法执行。

现在，Tornado 4.2.1 给了我“IOLoop 已经在运行”，但在 Tornado 3.1.1 中它可以工作（但我不知道具体如何）。

我知道接下来的事情：

我可以创建新的 ioloop，但我想使用现有的 ioloop。
我可以用一些知道 genlist 的结果是 Future 的函数来包装 genlist，但我不知道如何阻止执行，直到将 future 的结果设置在同步函数中。

另外，我不能使用 a.execute() 的结果作为未来对象，因为 a.execute() 可以从代码的其他部分调用，即它应该返回列表实例.

所以，我的问题是：是否有机会使用当前 IOLoop 从同步模型的方法执行异步 genlist？

【问题讨论】：

标签： python asynchronous tornado future coroutine

【解决方案1】：

您不能在此处重新启动外部 IOLoop。您有三个选择：

在任何地方都使用异步接口：将a.execute() 和堆栈顶部的所有内容更改为协程。这是基于 Tornado 的应用程序的常见模式；试图跨越同步和异步世界是很困难的，最好是站在一边或另一边。
在临时的IOLoop 上使用run_sync()。这就是 Tornado 的同步 tornado.httpclient.HTTPClient 所做的事情，这使得从另一个 IOLoop 中调用是安全的。但是，如果您这样做，外部 IOLoop 仍然被阻塞，因此您将 genlist 设为异步将一无所获。

在单独的线程上运行a.execute 并回调主IOLoop 的线程以获取内部函数。如果 a.execute 不能异步，这是避免 IOLoop 在运行时阻塞的唯一方法。

executor = concurrent.futures.ThreadPoolExecutor(8)

@gen.coroutine
def some_work():
    a = A()
    def adapter():
        # Convert the thread-unsafe tornado.concurrent.Future
        # to a thread-safe concurrent.futures.Future.
        # Note that everything including chain_future must happen
        # on the IOLoop thread.
        future = concurrent.futures.Future()
        ioloop.IOLoop.instance().add_callback(
            lambda: tornado.concurrent.chain_future(
                genlist(), future)
        return future.result()
    a.f = adapter
    print "a.f set"
    raise gen.Return(a)

@gen.coroutine
def main():
    a = yield some_work()
    retval = yield executor.submit(a.execute)
    raise gen.Return(retval)

【讨论】：

【解决方案2】：

说，你的函数看起来像这样：

@gen.coroutine
def foo():
    # does slow things

或

@concurrent.run_on_executor
def bar(i=1):
    # does slow things

你可以像这样运行foo()：

from tornado.ioloop import IOLoop
loop = IOLoop.current()

loop.run_sync(foo)

你可以运行bar(..)，或者任何需要参数的协程，像这样：

from functools import partial
from tornado.ioloop import IOLoop

loop = IOLoop.current()

f = partial(bar, i=100)
loop.run_sync(f)

【讨论】：