python - gevent 到 Tornado ioloop - 带有协程/生成器的结构代码

Question

我正在尝试将一些相当简单的 gevent 代码转换为使用 Tornado 的异步功能。下面的示例代码使用 ZMQ 库来执行非常简单的请求-响应。

import zmq.green as zmq

def fun():
    i = zmq.Context.instance()
    sock = i.socket(zmq.REQ)
    sock.connect('tcp://localhost:9005')
    sock.send('Ping')
    return sock.recv()

我可以在我的代码中的任何地方以 fun() 的形式运行它。 .recv() 调用在等待回复时阻塞，gevent 中心可以安排代码的其他部分。当收到值时，该函数返回该值。

我读了problems that can arise with these implicit returns ，我想使用 Tornado IOLoop 运行它(也是因为我想在 IPython Notebook 中运行它)。以下是一个选项，其中 recv_future() 返回包含结果的 Future:

@gen.coroutine
def fun():
    i = zmq.Context.instance()
    sock = i.socket(zmq.REQ)
    sock.connect('tcp://localhost:9005')
    sock.send('Ping')
    msg = yield recv_future(sock)
    print "Received {}".format(msg[0])
    raise gen.Return(msg)

def recv_future(socket):
    zmqstream = ZMQStream(socket)  # Required for ZMQ
    future = Future()
    def _finish(reply):
        future.set_result(reply)
    zmqstream.on_recv(_finish)
    return future

问题是现在 fun() 不是一个函数，而是一个生成器。所以如果我需要从另一个函数调用它，我需要使用 yield fun()。但是调用函数也变成了一个生成器!

构建使用 Python 生成器的代码的正确方法是什么？我是否必须让每个函数都成为生成器才能使其工作？如果我需要从 __init__() 调用其中一个函数怎么办？那也应该成为发电机吗？

Answer 1

What if I need to call one of these functions from __init__()? Should that also become a generator?

这是使用 yield/yield from(在 Python 3.3+ 上)的显式异步编程当前 Unresolved 问题之一。魔术方法不支持它们。您可以阅读 Python 核心开发人员关于涉及此问题的异步编程的一些有趣想法 here .

What is the right way to structure code that uses Python generators? Do I have to make every function a generator to make it work? Not every function, but every function that you want to call a coroutine, and wait for that coroutine to finish before continuing. When you switch to an explicit asynchronous programming model, you generally want to go all-in with it - your entire program runs inside the tornado ioloop. So, with this toy example, you would just do:

from tornado.ioloop import IOLoop
from tornado.gen import coroutine
from tornado.concurrent import Future

@gen.coroutine
def fun():
    i = zmq.Context.instance()
    sock = i.socket(zmq.REQ)
    sock.connect('tcp://localhost:9005')
    sock.send('Ping')
    msg = yield recv_future(sock)
    print "Received {}".format(msg[0])
    raise gen.Return(msg)

def recv_future(socket):
    zmqstream = ZMQStream(socket)  # Required for ZMQ
    future = Future()
    def _finish(reply):
        future.set_result(reply)
    zmqstream.on_recv(_finish)
    return future

if __name__ == "__main__":
    ioloop = IOLoop.instance()
    ioloop.add_callback(fun)
    ioloop.start() # This will run fun, and then block forever.
    #ioloop.run_sync(fun) # This will start the ioloop, run fun, then stop the ioloop

看来您可以通过 IPython.kernel API 访问 IPython 正在使用的 ioloop :

In [4]: from IPython.kernel.ioloop import manager

In [5]: manager.ioloop.IOLoop.instance()
Out[5]: <zmq.eventloop.ioloop.ZMQIOLoop at 0x4249ac8>