tornado并行执行多个异步请求

水深无声 2022-05-27 04:07 439阅读 0赞

Tornado实现后台逻辑的时候，可能遇到这样的情况： 需要同时请求多个第三方数据，比如同时从多个网址请求数据，而这几个第三方数据相互没有关联。 最简单的方式是写多个yield，第一个yield返回结果之后，继续请求第二个yield。这样虽然不会影响总体的性能，因为当前yield进行的时候，程序可以继续执行其他的请求，而无需等待在这边。 但是对于单个的请求来讲， 从它的视角来看， 就是顺序的请求多个第三方数据，然后返回，这样会造成这个请求的处理时间比较长。

那么Tornado有没有这种机制，对于单个请求来讲，如果需要请求多个无关联的外部数据，可以同时将这几个数据请求发出，等所有的数据返回之后再进行下面的流程。 相当于小学时的数学题，刷牙需要2分钟， 泡面需要3分钟， 我们无需刷完牙再泡面；而是泡面的同时可以刷牙，这样可以将单个流程的5分钟减少到3分钟。当然答案是肯定的， Tornado有类似的机制。

tornado可以用如下方式，同时并发n个请求：

response1, response2,... responsen = yield \[http\_client.fetch(url1) , http\_client.fetch(url2), ...... ,http\_client.fetch(url2) \]

等到n个请求都响应了之后，会返回给程序控制权

对于其中的原理，在网上找到一篇文章，讲得很详细，

下面的都是转载的内容， 原始文章地址： http://www.pulpcode.cn/2016/03/06/tornado-yield-futures-run-in-parallel/

##  ##

# tornado并行执行多个异步的原理 #

## 起因 ##

实际上之前一直使用tornado异步，也大概知道tornado Coroutine yield的原理，从来不知道tornado可以支持“同时多个异步并发执行，等它们的结果都获取到再返回”，官网中给出类似的写法：

You can also yield a list or dict of Futures, which will be started at the same time and run in parallel; a list or dict of results will be returned when they are all finished:

@gen.coroutine
    def get(self):
    http_client = AsyncHTTPClient()
    response1, response2 = yield [http_client.fetch(url1),
    http_client.fetch(url2)]
    response_dict = yield dict(response3=http_client.fetch(url3),
    response4=http_client.fetch(url4))
    response3 = response_dict['response3']
    response4 = response_dict['response4']

所以这篇博客就来挖掘一下原因。

## 前提知识 ##

如果你想看懂这篇文章，你要知道什么是yield，什么是异步，还有什么是协程，对tornado异步的原理了解，里面用到的Futrue对象,coroutine之类,它们是如何被tornado的ioloop进行上下文切换的。

当你在一个函数中使用yield的时候，这个函数就被称为一个生成器，这和普通函数不同的是，它可以被中断，什么是中断呢？比如现在有一个函数，你去调用它，这个时候，你就只能在这里等调用结果，什么都不能干，因为控制权已经交给函数了，当然比这高级一点的是将回调函数传入这个函数，使得这个函数能够在某些情况发生时，调用这个回调函数。

而有了yield则更高级，可以在这个函数执行的途中，将控制权在返还给调用者，这样我们可以在做一些事情之后，继续让这个函数执行下去。这才是yield真正的精髓所在，比xrange要高端多了。

#### 传递参数 ####

如果一个函数中有一个这样的一条语句：

`m = yield n`,那其实意味着，在函数内部与外部的沟通中，n将会被生成器送出，m将会被外部调用者传入（通过send）。

对应的tornado这条语句：`response = yield http_client.fetch(url)`

实际上，http\_client.fetch(url)返回了一个Future对象，当这个handler函数(比如get)被装饰器包裹之后,它就会通过generator.next()启动yield返回的generator,通过ioloop与generator.send(vaule)驱动generator运行,已达到异步执行的目的。

而在tornado的coroutine异步处理中,都是通过Future对象封装异步回调函数的。你看见Futrue，一定会想起，python3.2新增的concurrent.futures功能，实际上功能确实类似，或者说思想上是一样的，既“封装callable，以便异步执行。”你可以简单的记为:tornado异步函数将返回一个Future对象，yield这个对象，将得到最终结果。

## Future对象 ##

这个Future对象有一个叫\_set\_done的flag，当调用set\_result(self,result)来为这个Future对象设置result时会设置\_set\_done。在设置set\_done之后，它所有的add\_done\_callback才会被执行。之后你就可以通过result方法，来获取最终结果。

Future的一个简单的例子：

class HelloHandler(RequestHandler):
    @gen.coroutine
    def get(self):
    x = yield self.do_test()
    self.render("xxxx")
    def do_test(self):
    fut = Future()
    fut.set_result("test")
    return fut

## 寻找问题 ##

讲完上面的基础知识，我们该去找原问题了，既“同时多个异步并发执行，等它们的结果都获取到再返回”的支持代码.

首先我们找到tornado的源码，找到`coroutine`其实是这样实现的。

def _make_coroutine_wrapper(func, replace_callback):
    """The inner workings of ``@gen.coroutine`` and ``@gen.engine``.
    The two decorators differ in their treatment of the ``callback``
    argument, so we cannot simply implement ``@engine`` in terms of
    ``@coroutine``.
    """
    # On Python 3.5, set the coroutine flag on our generator, to allow it
    # to be used with 'await'.
    if hasattr(types, 'coroutine'):
    func = types.coroutine(func)
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
    future = TracebackFuture()
    if replace_callback and 'callback' in kwargs:
    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(sys.exc_info())
    return future
    else:
    if isinstance(result, GeneratorType):# 当它是生成器的时候。
    # Inline the first iteration of Runner.run. This lets us
    # avoid the cost of creating a Runner when the coroutine
    # never actually yields, which in turn allows us to
    # use "optional" coroutines in critical path code without
    # performance penalty for the synchronous case.
    try:
    orig_stack_contexts = stack_context._state.contexts
    yielded = next(result) # result作为一个生成器将会执行到yield处，并返回一个Future对象。
    if stack_context._state.contexts is not orig_stack_contexts:
    yielded = TracebackFuture()
    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(_value_from_stopiteration(e))
    except Exception:
    future.set_exc_info(sys.exc_info())
    else:
    Runner(result, future, yielded)
    try:
    return future
    finally:
    # Subtle memory optimization: if next() raised an exception,
    # the future's exc_info contains a traceback which
    # includes this stack frame. This creates a cycle,
    # which will be collected at the next full GC but has
    # been shown to greatly increase memory usage of
    # benchmarks (relative to the refcount-based scheme
    # used in the absence of cycles). We can avoid the
    # cycle by clearing the local variable after we return it.
    future = None
    future.set_result(result)
    return future
    return wrapper

可以看到关键的这几句:

<table style="border-spacing:0px;width:auto;border:none;font-size:14px;margin:0px;"> 
 <tbody> 
  <tr style="background-color:rgb(249,249,249);"> 
   <td style="padding:0px;vertical-align:middle;border-style:none;"><pre style="font-family:'PT Mono', Consolas, Monaco, Menlo, monospace;font-size:13px;background:rgb(247,247,247);margin-bottom:0px;padding-right:20px;padding-left:0px;color:rgb(102,102,102);line-height:1.6;border:none;text-align:right;"></pre> 
    <div style="height:20px;">
      1 
    </div> 
    <div style="height:20px;">
      2 
    </div> 
    <div style="height:20px;">
      3 
    </div></td> 
   <td style="padding:0px;vertical-align:middle;border-style:none;"><pre style="font-family:'PT Mono', Consolas, Monaco, Menlo, monospace;font-size:13px;background:rgb(247,247,247);margin-bottom:0px;padding-right:0px;padding-left:0px;color:rgb(77,77,76);line-height:1.6;border:none;"></pre> 
    <div style="height:20px;">
      result = func(*args, **kwargs) 
    </div> 
    <div style="height:20px;">
      yielded = next(result) 
    </div> 
    <div style="height:20px;">
      Runner(result, future, yielded) 
    </div></td> 
  </tr> 
 </tbody> 
</table>

简单的说，这个函数，捕获被装饰器函数返回的生成器对象，传递给Runner。

Runner的代码实际上是这样的：

def __init__(self, gen, result_future, first_yielded):
    self.future = first_yield
    self.io_loop.add_future(
    self.future, lambda f: self.run()
    )
    def run(self):
    while True:
    if not future.done():
    return
    try:
    value = future.result()
    yielded = self.gen.send(value)
    except (StopIteration, Return) as e:
    self.finished = True
    except Exception:
    self.finished = True
    return
    if not self.handle_yield(yielded):
    return

而这个Runner，会将Futrue对象注册到io\_loop中，或者就之前的例子，我们可以说将异步函数fetch注册到ioloop，当fetch完成后，它会调用自己的一个回调函数（我们这里讨论的是没有给fetch传递callback的情况，详见AsyncHTTPClient的定义），给future对象设置值。而io\_loop又将会调用回调函数`lambda f: self.run()`,来将future.result的值赋值给value.  
可以看到这个value被send给了生成器内部，之后生成器又会得到下一个执行点（生成器）

这个runner使用迭代的方式，一个一个的获取生成器，处理完之后，就使用handle\_yield，判断是否要还有下一个Future对象和回调，（之所以用迭代，而不是递归，是因为在python中递归是一件很慢的事儿。）直到所有的future都done了。

接着我们将注意力放到这个handle\_yield函数中。

def handle_yield(self, yielded):
    # Lists containing YieldPoints require stack contexts;
    # other lists are handled in convert_yielded.
    if _contains_yieldpoint(yielded):
    yielded = multi(yielded)
    if isinstance(yielded, YieldPoint):
    # YieldPoints are too closely coupled to the Runner to go
    # through the generic convert_yielded mechanism.
    self.future = TracebackFuture()
    def start_yield_point():
    try:
    yielded.start(self)
    if yielded.is_ready():
    self.future.set_result(
    yielded.get_result())
    else:
    self.yield_point = yielded
    except Exception:
    self.future = TracebackFuture()
    self.future.set_exc_info(sys.exc_info())
    if self.stack_context_deactivate is None:
    # Start a stack context if this is the first
    # YieldPoint we've seen.
    with stack_context.ExceptionStackContext(
    self.handle_exception) as deactivate:
    self.stack_context_deactivate = deactivate
    def cb():
    start_yield_point()
    self.run()
    self.io_loop.add_callback(cb)
    return False
    else:
    start_yield_point()
    else:
    try:
    self.future = convert_yielded(yielded)
    except BadYieldError:
    self.future = TracebackFuture()
    self.future.set_exc_info(sys.exc_info())
    if not self.future.done() or self.future is moment:
    self.io_loop.add_future(
    self.future, lambda f: self.run())
    return False
    return True

可以看到

if not self.future.done() or self.future is moment:
    self.io_loop.add_future(
    self.future, lambda f: self.run())

所以这就是循环的去执行到下一个yield的方式。还要注意的是，代码中提到的YieldPoint已经被放弃，Tornado4.0也推荐使用Future类型。

再注意convert\_yielded函数。

def convert_yielded(yielded):
    """Convert a yielded object into a `.Future`.
    The default implementation accepts lists, dictionaries, and Futures.
    If the `~functools.singledispatch` library is available, this function
    may be extended to support additional types. For example::
    @convert_yielded.register(asyncio.Future)
    def _(asyncio_future):
    return tornado.platform.asyncio.to_tornado_future(asyncio_future)
    .. versionadded:: 4.1
    """
    # Lists and dicts containing YieldPoints were handled earlier.
    if isinstance(yielded, (list, dict)):
    return multi(yielded)
    elif is_future(yielded):
    return yielded
    elif isawaitable(yielded):
    return _wrap_awaitable(yielded)
    else:
    raise BadYieldError("yielded unknown object %r" % (yielded,))

我们注意到了multi(yielded)这个调用，最后我们找到了这段代码，这就是答案：

def multi(children, quiet_exceptions=()):
    """Runs multiple asynchronous operations in parallel.
    ``children`` may either be a list or a dict whose values are
    yieldable objects. ``multi()`` returns a new yieldable
    object that resolves to a parallel structure containing their
    results. If ``children`` is a list, the result is a list of
    results in the same order; if it is a dict, the result is a dict
    with the same keys.
    That is, ``results = yield multi(list_of_futures)`` is equivalent
    to::
    results = []
    for future in list_of_futures:
    results.append(yield future)
    If any children raise exceptions, ``multi()`` will raise the first
    one. All others will be logged, unless they are of types
    contained in the ``quiet_exceptions`` argument.
    If any of the inputs are `YieldPoints <YieldPoint>`, the returned
    yieldable object is a `YieldPoint`. Otherwise, returns a `.Future`.
    This means that the result of `multi` can be used in a native
    coroutine if and only if all of its children can be.
    In a ``yield``-based coroutine, it is not normally necessary to
    call this function directly, since the coroutine runner will
    do it automatically when a list or dict is yielded. However,
    it is necessary in ``await``-based coroutines, or to pass
    the ``quiet_exceptions`` argument.
    This function is available under the names ``multi()`` and ``Multi()``
    for historical reasons.
    .. versionchanged:: 4.2
    If multiple yieldables fail, any exceptions after the first
    (which is raised) will be logged. Added the ``quiet_exceptions``
    argument to suppress this logging for selected exception types.
    .. versionchanged:: 4.3
    Replaced the class ``Multi`` and the function ``multi_future``
    with a unified function ``multi``. Added support for yieldables
    other than `YieldPoint` and `.Future`.
    """
    if _contains_yieldpoint(children):
    return MultiYieldPoint(children, quiet_exceptions=quiet_exceptions)
    else:
    return multi_future(children, quiet_exceptions=quiet_exceptions)

之后是multi\_future的定义：

def multi_future(children, quiet_exceptions=()):
    """Wait for multiple asynchronous futures in parallel.
    This function is similar to `multi`, but does not support
    `YieldPoints <YieldPoint>`.
    .. versionadded:: 4.0
    .. versionchanged:: 4.2
    If multiple ``Futures`` fail, any exceptions after the first (which is
    raised) will be logged. Added the ``quiet_exceptions``
    argument to suppress this logging for selected exception types.
    .. deprecated:: 4.3
    Use `multi` instead.
    """
    if isinstance(children, dict):
    keys = list(children.keys())
    children = children.values()
    else:
    keys = None
    children = list(map(convert_yielded, children))
    assert all(is_future(i) for i in children)
    unfinished_children = set(children)
    future = Future()
    if not children:
    future.set_result({} if keys is not None else [])
    def callback(f):
    unfinished_children.remove(f)
    if not unfinished_children:
    result_list = []
    for f in children:
    try:
    result_list.append(f.result())
    except Exception as e:
    if future.done():
    if not isinstance(e, quiet_exceptions):
    app_log.error("Multiple exceptions in yield list",
    exc_info=True)
    else:
    future.set_exc_info(sys.exc_info())
    if not future.done():
    if keys is not None:
    future.set_result(dict(zip(keys, result_list)))
    else:
    future.set_result(result_list)
    listening = set()
    for f in children:
    if f not in listening:
    listening.add(f)
    f.add_done_callback(callback)
    return future

这是它支持并行异步的关键代码，可以看到这个被包装的Future,它的listening,维护着多个子future，每次有一个子future完成后，就会调用callback，将其从unfinished\_children中移除，当所有子Future的callback都执行完后，才会真正调用这个Future的set\_result方法。