python Tornado异步使用场景源码解析
作者：NodeKey
前言
tornado在我工作前说实话我还没有听说过，今年回来后接触的非常多。
关于想要学习异步的内容，起因是如下场景：
tornado拿到请求后，对他的处理时间非常的长，一般大概有10s，晨旭让我把他写成异步的，如果只是对请求异步的是相当好写的。就是十分传统地在收到请求后立即返回，然后进行处理，处理了之后再返回给给定的callback_url。
但我突然想到，能否对这整个处理进行异步，将其放到后台运行，然后继续接收请求不至于在请求上堵塞。
最后是没实现出来……坤爷让我去花时间了解一下tornado所著名的异步。于是我才发现，我这想法在tornado的异步中是不可行的。(而且错误的地方还蛮多的……
异步使用方式
from tornado import gen
@gen.coroutine
def fetch_coroutine(url):
http_client = AsyncHTTPClient()
response = yield http_client.fetch(url)
raise gen.Return(response.body)这是最常用也是tornado推荐的方式。不过这样写的前提是py2，py3有些新特性，写的方式就不太一样了。
正如代码中展示的，你需要给这个函数加个gen.coroutine装饰器，然后在你想要异步处理的函数加个yield。
说实话，这已经有点跟我想的不太一样了，callback_function呢？没了？emmmmm好吧，这样写的确是没有的。
关于协程
简单介绍以下协程，协程是线程通过自己来调度任务，保存恢复上下文，而这样做的好处就是能减少切换线程或者进程的损耗，如果线程数量极多，但是做的事情不多，简单说运行类型是io密集型的话，可以考虑使用协程。
因为如果是cpu密集型的话，毕竟本质上还是一个线程，所以会堵塞到其他的协程，这与我们的高效目的是相违背的。
另外正式开始讲之前，我们首先需要明确的是一个函数如果带有yeild，那么这个函数实质上就是一个生成器。我们直接对其调用它返回的也是一个生成器。所有会有这样一句话：
所有生成器都是异步的
实际上也确实如此，生成器立即返回，我们想要执行器内容的话，可以自行调用next或者send(非第一次)
于是python的yield对于协程有了天然的支持
源码解析
注：本文章分析的源码是 2018.12.24 的 stable 版本
coroutine
协程，也是 tornado.gen 中的一个装饰器
这个装饰器其内容只有一个 return。而在 tornado.gen 中其实还有个装饰器engine，它的replace_callback 的 默认值为False
return _make_coroutine_wrapper(func, replace_callback=True)
以下是_make_coroutine_wrapper 的源码，源码十分紧凑，考虑的很全面，先说下大多数情况
def _make_coroutine_wrapper(func, replace_callback):
wrapped = func
if hasattr(types, 'coroutine'):
func = types.coroutine(func)
@functools.wraps(wrapped)
def wrapper(*args, **kwargs):
future = _create_future()
if replace_callback and 'callback' in kwargs:
warnings.warn("callback arguments are deprecated, use the returned Future instead",
DeprecationWarning, stacklevel=2)
callback = kwargs.pop('callback')
IOLoop.current().add_future(
future, lambda future: callback(future.result()))
try:
result = func(*args, **kwargs)
except (Return, StopIteration) as e:
result = _value_from_stopiteration(e)
except Exception:
future_set_exc_info(future, sys.exc_info())
try:
return future
finally:
# Avoid circular references
future = None
else:
if isinstance(result, GeneratorType):
try:
orig_stack_contexts = stack_context._state.contexts
yielded = next(result)
if stack_context._state.contexts is not orig_stack_contexts:
yielded = _create_future()
yielded.set_exception(
stack_context.StackContextInconsistentError(
'stack_context inconsistency (probably caused '
'by yield within a "with StackContext" block)'))
except (StopIteration, Return) as e:
future_set_result_unless_cancelled(future, _value_from_stopiteration(e))
except Exception:
future_set_exc_info(future, sys.exc_info())
else:
runner = Runner(result, future, yielded)
# 这一行不太理解，注释是说让它保持存在
future.add_done_callback(lambda _: runner)
yielded = None
try:
return future
finally:
future = None
future_set_result_unless_cancelled(future, result)
return future
wrapper.__wrapped__ = wrapped
wrapper.__tornado_coroutine__ = True
return wrappercoroutine 装饰的函数，首先会生成一个 generator, 并对其第一次 next 调用。
这里我们可以注意到，这个yield 后的函数无论如何它都是会被立刻调用的。
所以 yield 后的函数必须也是异步或者耗时不长的，才能达到预期的异步效果，否则该阻塞的还是会阻塞。
在 next调用后，有个异常捕获，在代码 36L，这个异常捕获犹为重要，因为我们知道 yield 等式，比如var = yield func()，var不会被赋值成func的返回。tornado提供了一个异常类Return 作为返回，通过在调用出捕获Return异常，取出其返回值得以实现。
在第一次的next后，如果没有其他异常，就会创建一个Runner类，这个Runner类的作用就是，把其他的代码通过yield不断暂停恢复，放在ioloop里运行。
Future
Future可以说是个中介，它是用来沟通coroutine和ioloop的，coroutine返回的都是Future
但其实最重要的还是管理协程的暂停与恢复，一个ioloop中保存着多个后端运行类Runner类的runner方法，在ioloop中不断暂停恢复，而每一个runner又都会绑定一个future，只有future被set_done了，才表示上一阶段已经完成并暂停了，可以继续恢复运行。
class Future(object):
def done(self):# 协程执行完毕并暂停，可对其恢复
return self._done
def result(self, timeout=None):
self._check_done() # 如果没有 done 抛出异常
return self._result
def add_done_callback(self, fn): # 添加回调函数
if self._done:
from tornado.ioloop import IOLoop
IOLoop.current().add_callback(fn, self)
else:
self._callbacks.append(fn)
def set_result(self, result): # 设置result & done
self._result = result
self._set_done()
def _set_done(self): # 将所有回调函数放到 ioloop中
self._done = True
if self._callbacks:
from tornado.ioloop import IOLoop
loop = IOLoop.current()
for cb in self._callbacks:
loop.add_callback(cb, self)
self._callbacks = NoneIOLoop
IOLoop是在整个tornado的主事件循环。按我理解主要做了两件事
- 执行异步的callback
- io复用
并且这两件事它是写死的，它是顺序执行的，这就直接反驳了我最开始的想法：让两者并行执行
io复用根据系统会自动调整的，具体的我也不再细说。
以下是精简之后的源码
class PollIOLoop(IOLoop):
def add_future(self, future, callback):
assert is_future(future)
callback = stack_context.wrap(callback)
future_add_done_callback(
future, lambda future: self.add_callback(callback, future))
def add_callback(self, callback, *args, **kwargs):
if self._closing:
return
self._callbacks.append(functools.partial(
stack_context.wrap(callback), *args, **kwargs)) # 将其以偏函数形式保存起来
if thread.get_ident() != self._thread_ident:
self._waker.wake()
else:
pass
def start(self):
# 这里有一堆初始化操作
while True:
ncallbacks = len(self._callbacks)
due_timeouts = []
if self._timeouts:
now = self.time()
while self._timeouts:
if self._timeouts[0].callback is None:
heapq.heappop(self._timeouts)
self._cancellations -= 1
elif self._timeouts[0].deadline <= now:
due_timeouts.append(heapq.heappop(self._timeouts))
else:
break
if (self._cancellations > 512 and
self._cancellations > (len(self._timeouts) >> 1)):
self._cancellations = 0
self._timeouts = [x for x in self._timeouts
if x.callback is not None]
heapq.heapify(self._timeouts)
for i in range(ncallbacks):
self._run_callback(self._callbacks.popleft())
for timeout in due_timeouts:
if timeout.callback is not None:
self._run_callback(timeout.callback)
if not self._running:
break
# 这里有设置poll_timeout
event_pairs = self._impl.poll(poll_timeout)
self._events.update(event_pairs)
while self._events:
fd, events = self._events.popitem()
try:
fd_obj, handler_func = self._handlers[fd]
handler_func(fd_obj, events)
except (OSError, IOError) as e:
if errno_from_exception(e) == errno.EPIPE:
pass
else:
self.handle_callback_exception(self._handlers.get(fd))
except Exception:
self.handle_callback_exception(self._handlers.get(fd))
fd_obj = handler_func = None
# 这里有一些后置处理Runner
Runner.run是后置处理的主函数，我在之前也有提到过，它通过获取future的result，再将其send以恢复协程继续执行。
如果不能捕获到任何异常，就说明有新的coroutine，新的coroutine都是通过handle_yield将其放进ioloop
class Runner(object):
def __init__(self, gen, result_future, first_yielded):
self.gen = gen
self.result_future = result_future
self.future = _null_future
self.yield_point = None
self.pending_callbacks = None
self.results = None
self.running = False
self.finished = False
self.had_exception = False
self.io_loop = IOLoop.current()
self.stack_context_deactivate = None
# 上面一堆不需要看的初始化
if self.handle_yield(first_yielded):
gen = result_future = first_yielded = None
self.run()
def handle_yield(self, yielded):
self.future = convert_yielded(yielded)
if self.future is moment:
self.io_loop.add_callback(self.run)
return False
elif not self.future.done():
def inner(f):
# Break a reference cycle to speed GC.
f = None
self.run()
self.io_loop.add_future(
self.future, inner)
return False
return True
def run(self):
if self.running or self.finished:
return
try:
self.running = True
while True:
future = self.future
if not future.done():
return
self.future = None
try:
orig_stack_contexts = stack_context._state.contexts
exc_info = None
try:
value = future.result()
except Exception:
self.had_exception = True
exc_info = sys.exc_info()
future = None
yielded = self.gen.send(value)
except (StopIteration, Return) as e:
self.finished = True
self.future = _null_future
if self.pending_callbacks and not self.had_exception:
raise LeakedCallbackError(
"finished without waiting for callbacks %r" %
self.pending_callbacks)
future_set_result_unless_cancelled(self.result_future,
_value_from_stopiteration(e))
self.result_future = None
self._deactivate_stack_context()
return
except Exception:
# 一些结束操作
return
if not self.handle_yield(yielded):
return
yielded = None
finally:
self.running = False以上就是python Tornado异步使用场景源码解析的详细内容，更多关于python Tornado异步的资料请关注脚本之家其它相关文章！