概览

场景
实现
方法
总结
1. 场景

在写异步爬虫时，发现很多请求莫名其妙地超时。

原来是因为解析网页耗费了太多时间，使得部分请求超过预定时间。

根本上是因为，asyncio 的协程是非抢占式的。协程如果不主动交出控制权，就会一直执行下去。

假如一个协程占用了太多时间，那么其他协程就有可能超时挂掉。

下面我们用一个简短的程序实验一下。

import asyncio
import time


async def long_calc():
    print('long calc start')
    time.sleep(3)
    print('long calc end')


async def waiting_task(i):
    print(f'waiting task {i} start')
    try:
        await asyncio.wait_for(asyncio.sleep(1), 1.5)
    except asyncio.TimeoutError:
        print(f'waiting task {i} timeout')
    else:
        print(f'waiting task {i} end')


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    coros = [long_calc()]
    coros.extend(waiting_task(x) for x in range(3))
    loop.run_until_complete(asyncio.wait(coros))

long_calc 模拟高耗时的同步处理；
waiting_task 模拟有时间限制的异步处理，时间设为 1 秒，时限设为 1.5 秒。

运行三个异步，一个同步，结果是这样的：

waiting task 0 start
waiting task 1 start
long calc start
long calc end
waiting task 2 start
waiting task 0 timeout
waiting task 1 timeout
waiting task 2 end

在 long_calc 启动前的 waiting_task 超时报错，long_calc 完成后的 waiting_task 正常结束。

很明显，开过多同步任务和这种情况是一样的。

在异步爬虫的场景来说，就是过多或过长的解析让请求超时。

合理的架构设计当然能避免这个冲突，这里提供另外的解决办法。

2. 实现

话不多说，先看代码：

import asyncio
import time
import threading
import datetime


def print_with_time(fmt):
    print(str(datetime.datetime.now()) + ': ' + fmt)


async def long_calc():
    print_with_time('long calc start')
    time.sleep(8)
    print_with_time('long calc end')


async def waiting_task(i):
    print_with_time(f'waiting task {i} start')
    try:
        await asyncio.wait_for(asyncio.sleep(3), 5)
    except asyncio.TimeoutError:
        print_with_time(f'waiting task {i} timeout')
    else:
        print_with_time(f'waiting task {i} end')


if __name__ == '__main__':
    sub_loop = asyncio.new_event_loop()
    thread = threading.Thread(target=sub_loop.run_forever)
    thread.start()

    loop = asyncio.get_event_loop()
    task = loop.create_task(long_calc())
    futs = [asyncio.run_coroutine_threadsafe(waiting_task(x), loop=sub_loop) for x in range(3)]
    futs = [asyncio.wrap_future(f, loop=loop) for f in futs]

    loop.run_until_complete(asyncio.wait([task, *futs]))

    sub_loop.call_soon_threadsafe(sub_loop.stop)
    thread.join()

还以 long_calc 和 waiting_task 为例，时间调整了一下,同步耗时 8 秒,异步耗时 3 秒，时限 5 秒。

起一个线程，运行附属事件循环，并用 asyncio 中线程安全的方法添加异步任务。

我们将异步任务放在一个附属事件循环中运行，将同步任务放在主循环中运行。多线程由系统控制，是抢占式的。显然两个线程互不打扰，避免了超时问题。（虽然内存和 CPU 占用都要高一些）

由 run_until_complete 那一行看出，主循环能接收到附属循环中任务完成的信息。

最后把 stop 方法 call 进附属循环，等待子线程结束。

运行结果：

2018-06-26 21:37:50.957276: waiting task 0 start
2018-06-26 21:37:50.958276: long calc start
2018-06-26 21:37:50.958276: waiting task 1 start
2018-06-26 21:37:50.958276: waiting task 2 start
2018-06-26 21:37:53.958448: waiting task 0 end
2018-06-26 21:37:53.958448: waiting task 2 end
2018-06-26 21:37:53.958448: waiting task 1 end
2018-06-26 21:37:58.958734: long calc end

完美协调。

3. 方法

这种实现基于 asyncio 的两个方法：

def run_coroutine_threadsafe(coro, loop)

def wrap_future(future, *, loop=None)

先看第一个：

def run_coroutine_threadsafe(coro, loop):
    """Submit a coroutine object to a given event loop.

    Return a concurrent.futures.Future to access the result.
    """
    if not coroutines.iscoroutine(coro):
        raise TypeError('A coroutine object is required')
    future = concurrent.futures.Future()

    def callback():
        try:
            futures._chain_future(ensure_future(coro, loop=loop), future)
        except Exception as exc:
            if future.set_running_or_notify_cancel():
                future.set_exception(exc)
            raise

    loop.call_soon_threadsafe(callback)
    return future

向给定的事件循环提交一个协程，并且保证线程安全。

第一步： 类型检验（Python 之所以慢。。。）

第二步： 新建一个 Future。这个 Future 和 asyncio 的 Future 不同，是线程安全的，也用于 concurrent 的 executor 实现。

第三步： 闭包函数，将协程包装为循环的 future，再与线程安全的 future 连接。（???）

第四步： 把闭包函数 call 进循环，返回线程安全的 future。

有疑点，暂时放着。看下一个：

def wrap_future(future, *, loop=None):
    """Wrap concurrent.futures.Future object."""
    if isfuture(future):
        return future
    assert isinstance(future, concurrent.futures.Future), \
        'concurrent.futures.Future is expected, got {!r}'.format(future)
    if loop is None:
        loop = events.get_event_loop()
    new_future = loop.create_future()
    _chain_future(future, new_future)
    return new_future

类型检验，确定循环，创建循环的 future，连接两个 future (???)，返回新的 future。

我们再次见到了_chain_future，那么看看它的源码：

def _chain_future(source, destination):
    """Chain two futures so that when one completes, so does the other.

    The result (or exception) of source will be copied to destination.
    If destination is cancelled, source gets cancelled too.
    Compatible with both asyncio.Future and concurrent.futures.Future.
    """
    if not isfuture(source) and not isinstance(source,
                                               concurrent.futures.Future):
        raise TypeError('A future is required for source argument')
    if not isfuture(destination) and not isinstance(destination,
                                                    concurrent.futures.Future):
        raise TypeError('A future is required for destination argument')
    source_loop = source._loop if isfuture(source) else None
    dest_loop = destination._loop if isfuture(destination) else None

    def _set_state(future, other):
        if isfuture(future):
            _copy_future_state(other, future)
        else:
            _set_concurrent_future_state(future, other)

    def _call_check_cancel(destination):
        if destination.cancelled():
            if source_loop is None or source_loop is dest_loop:
                source.cancel()
            else:
                source_loop.call_soon_threadsafe(source.cancel)

    def _call_set_state(source):
        if dest_loop is None or dest_loop is source_loop:
            _set_state(destination, source)
        else:
            dest_loop.call_soon_threadsafe(_set_state, destination, source)

    destination.add_done_callback(_call_check_cancel)
    source.add_done_callback(_call_set_state)

对两种 future 统一处理，使得当 src 完成时，dst 也完成，并且线程安全。

这样就清楚了。附属循环内有一个 future 表示异步任务的状态，主线程中留下一个线程安全的 future，与附属循环内的 future 相连，最后在主循环中创建一个 future，与线程安全的 future 相连。

我们可以在主循环中 await 这个 future，和在同一个循环中一样，只不过不能跨循环使用 lock 和 timeout 机制而已。

4. 总结

对于这种场景，大部分情况下用线程池 executor 是可行的。

但如果我想多级爬取，解析后发请求，得到结果后，加上上级页面的信息组装成一个 Item，就要调用多次函数，传一堆参数，无法利用协程同一作用域的特性。

使用上述方法，就能折腾出比较舒服的写法，而主循环+附属循环的框架是可以抽象并复用的。

当然，这不一定是生产中实用的方法，但弄清楚这个，你对 asyncio 的理解一定会加深。

Python学习交流群：834179111，群里有很多的学习资料。欢迎欢迎各位前来交流学习。