导航
用户指南
协程
Coroutines是在Tornado中编写异步代码的推荐方法。Coroutines使用Pythonawait
或yield
关键字来挂起和恢复执行而不是一系列回调(在gevent这样的框架中看到的协作轻量级线程有时也被称为协程,但在Tornado中所有协程都使用显式上下文切换并被称为异步函数)。
协程几乎和同步代码一样简单,而且没有线程那样的昂贵开销。它们还通过减少可能发生的上下文切换来简化并发。
例子:
async def fetch_coroutine(url):
http_client = AsyncHTTPClient()
response = await http_client.fetch(url)
return response.body
原生协程VS装饰器协程
Python 3.5引入了async和await关键字(使用这些关键字的函数也称为“native coroutines”)。 为了与旧版本的Python兼容,您可以使用tornado.gen.coroutine装饰器来使用“decorated”或“yield-based”的协程。
尽可能使用原生协程。 仅在需要与旧版本的Python兼容时才使用装饰器协程。Tornado文档中的示例通常使用原生形式。
两种形式之间的转换通常很简单:
# Decorated: # Native:
# Normal function declaration
# with decorator # "async def" keywords
@gen.coroutine
def a(): async def a():
# "yield" all async funcs # "await" all async funcs
b = yield c() b = await c()
# "return" and "yield"
# cannot be mixed in
# Python 2, so raise a
# special exception. # Return normally
raise gen.Return(b) return b
其它两种形式的协程区别:
– 原生协程通常更快。
– 原生协程可以使用async for
和async
语句,这使得某些模式更加简单。
– 除非yield
或await
它们,否则原生协程根本不会运行。装饰器协程一旦被调用就可以“在后台”开始运行。请注意,对于这两种协程,使用await
或yield
很重要,这样任何异常才能正常抛出。
– 装饰器协程与concurrent.futures包有额外的集成,允许直接生成executor.submi
的结果。对于原生协程,请改用IOLoop.run_in_executor
。
– 装饰器协程通过产生列表或字典来支持等待多个对象的一些简写。使用tornado.gen.multi在原生协程中执行此操作。
– 装饰器协程可以支持与其他软件包的集成,包括通过转换函数注册表的Twisted。要在原生协程中访问此功能,请使用tornado.gen.convert_yielded
。
– 装饰器协程总是返回一个Future对象。原生协程返回一个不是Future的等待对象。在Tornado中,两者大多可以互换。
工作原理
本节介绍装饰器协程的操作。原生协程在概念上是相似的,但由于与Python运行时的额外集成而稍微复杂一些。
包含yield
的函数是生成器。所有生成器都是异步的,在调用时,它们返回一个生成器对象而不是运行到完成。 @gen.coroutine
装饰器通过yield
表达式与生成器通信,并通过返回Future
与协程的调用者通信。
这是协程装饰器内循环的简化版本:
# Simplified inner loop of tornado.gen.Runner
def run(self):
# send(x) makes the current yield return x.
# It returns when the next yield is reached
future = self.gen.send(self.next)
def callback(f):
self.next = f.result()
self.run()
future.add_done_callback(callback)
装饰器从生成器接收Future
,等待(不阻塞)该Future
完成,然后“展开”Future
并将结果作为yield
表达式的结果发送回生成器。 大多数异步代码从不直接接触Future
类,除非立即将异步函数返回的Future
传递给yield
表达式。
如何调用一个协程
协程不会以正常方式抛出异常:它们抛出的任何异常都将被困在等待对象中,直到它被放弃为止。 这意味着以正确的方式调用协同程序很重要,否则您可能会发现未被注意到的错误:
async def divide(x, y):
return x / y
def bad_call():
# This should raise a ZeroDivisionError, but it won't because
# the coroutine is called incorrectly.
divide(1, 0)
在几乎所有情况下,任何调用协程的函数都必须是协程本身,并在调用中使用await
或yield
关键字。 当重写超类中定义的方法时,请查阅文档以查看是否允许协程(文档应该说方法“可能是协程”或“可能返回Future
”):
async def good_call():
# await will unwrap the object returned by divide() and raise
# the exception.
await divide(1, 0)
有时你可能想要“Fire and forget”一个协程而不等待它的结果。在这种情况下,建议使用IOLoop.spawn_callback
,这使得IOLoop
负责调用。 如果失败,IOLoop
将记录堆栈路径:
# The IOLoop will catch the exception and print a stack trace in
# the logs. Note that this doesn't look like a normal call, since
# we pass the function object to be called by the IOLoop.
IOLoop.current().spawn_callback(divide, 1, 0)
对于使用@gen.coroutine
的函数,建议以这种方式使用IOLoop.spawn_callback
,但是使用async def
的函数需要它(否则协程运行程序将无法启动)。
最后,在程序的顶层,如果IOLoop
尚未运行,您可以启动IOLoop
,运行协程,然后使用IOLoop.run_sync
方法停止IOLoop。 这通常用于启动面向批处理( batch-oriented)程序的main
函数:
# run_sync() doesn't take arguments, so we must wrap the
# call in a lambda.
IOLoop.current().run_sync(lambda: divide(1, 0))
协程模式
调用阻塞函数(Calling blocking functions)
从协程中调用一个阻塞函数的最简单的方法就是使用ThreadPoolExecutor
,返回一个其他协程兼容的Futures
对象:
async def call_blocking():
await IOLoop.current().run_in_executor(None, blocking_func, args)
并行(Parallelism)
multi
函数接受其值为Futures
的列表和dicts,并且并行等待所有这些Futures
:
from tornado.gen import multi
async def parallel_fetch(url1, url2):
resp1, resp2 = await multi([http_client.fetch(url1),
http_client.fetch(url2)])
async def parallel_fetch_many(urls):
responses = await multi ([http_client.fetch(url) for url in urls])
# responses is a list of HTTPResponses in the same order
async def parallel_fetch_dict(urls):
responses = await multi({url: http_client.fetch(url)
for url in urls})
# responses is a dict {url: HTTPResponse}
在装饰器协程中,可以直接yield
列表或字典:
@gen.coroutine
def parallel_fetch_decorated(url1, url2):
resp1, resp2 = yield [http_client.fetch(url1),
http_client.fetch(url2)]
交叉存取(Interleaving)
有时保存一个Future对象比立即yield它会更有用,以便你可以在等待之前开始开始另一个操作:
from tornado.gen import convert_yielded
async def get(self):
# convert_yielded() starts the native coroutine in the background.
# This is equivalent to asyncio.ensure_future() (both work in Tornado).
fetch_future = convert_yielded(self.fetch_next_chunk())
while True:
chunk = yield fetch_future
if chunk is None: break
self.write(chunk)
fetch_future = convert_yielded(self.fetch_next_chunk())
yield self.flush()
这对于装饰的协同程序来说更容易一些,因为它们在被调用时立即启动:
@gen.coroutine
def get(self):
fetch_future = self.fetch_next_chunk()
while True:
chunk = yield fetch_future
if chunk is None: break
self.write(chunk)
fetch_future = self.fetch_next_chunk()
yield self.flush()
循环(Looping)
在原生协程中,可以使用async for
。在旧版本的Python中,循环对于协程来说很棘手,因为无法在for
循环或while
循环的每次迭代中yield
并捕获yield
的结果。相反,您需要将循环条件与访问结果分开,如本例中的Motor
:
import motor
db = motor.MotorClient().test
@gen.coroutine
def loop_example(collection):
cursor = db.collection.find()
while (yield cursor.fetch_next):
doc = cursor.next_object()
后台运行(Running in the background)
PeriodicCallback
通常不与协同程序一起使用。相反,一个协程可以包含一个while True:
:循环并使用tornado.gen.sleep
:
async def minute_loop():
while True:
await do_something()
await gen.sleep(60)
# Coroutines that loop forever are generally started with
# spawn_callback().
IOLoop.current().spawn_callback(minute_loop)
有时可能需要更复杂的循环。 例如,前一个循环每60 + N
秒运行一次,其中N
是do_something()
的运行时间。 要完全每60秒运行一次,请使用上面的交叉存取:
async def minute_loop2():
while True:
nxt = gen.sleep(60) # Start the clock.
await do_something() # Run while the clock is ticking.
await nxt # Wait for the timer to run out.
上一篇:异步与非阻塞I/O