从头造轮子：python3 asyncio之 gather （3）-编程知识-白鹭情

书接上文，本文造第三个轮子，也是asyncio包里面非常常用的一个函式gather

● 相对于前两个函式，gather的使用频率更高，因为它支持多个协程任务“同时”执行
● 理解__await__ __iter__的使用
● 理解关键字async/await，async/await是3.5之后的语法，和yield/yield from异曲同工
● 今天的文章有点长，请大家耐心看完

组件	版本
python	3.7.7

先来看下官方gather的使用方法：

|># more main.py
import asyncio

async def hello():
    print('enter hello ...')
    return 'return hello ...'

async def world():
    print('enter world ...')
    return 'return world ...'

async def helloworld():
    print('enter helloworld')
    ret = await asyncio.gather(hello(), world())
    print('exit helloworld')
    return ret

if __name__ == "__main__":
    ret = asyncio.run(helloworld())
    print(ret)
    
|># python3 main.py
enter helloworld
enter hello ...
enter world ...
exit helloworld
['return hello ...', 'return world ...']

来看下造的轮子的使用方式：

? more main.py
import wilsonasyncio

async def hello():
    print('enter hello ...')
    return 'return hello ...'

async def world():
    print('enter world ...')
    return 'return world ...'

async def helloworld():
    print('enter helloworld')
    ret = await wilsonasyncio.gather(hello(), world())
    print('exit helloworld')
    return ret


if __name__ == "__main__":
    ret = wilsonasyncio.run(helloworld())
    print(ret)

    
? python3 main.py
enter helloworld
enter hello ...
enter world ...
exit helloworld
['return hello ...', 'return world ...']

自己造的轮子也很好的运行了，下面我们来看下轮子的代码

轮子代码

1）代码组成

|># tree
.
├── eventloops.py 
├── futures.py
├── main.py
├── tasks.py
├── wilsonasyncio.py

档案	作用
eventloops.py	事件回圈
futures.py	futures物件
tasks.py	tasks物件
wilsonasyncio.py	可呼叫方法集合
main.py	入口

2）代码概览：

eventloops.py

类/函式	方法	物件	作用	描述
Eventloop			事件回圈，一个执行绪只有运行一个
	`__init__`		初始化两个重要物件 `self._ready` 与 `self._stopping`
		`self._ready`	所有的待执行任务都是从这个队列取出来，非常重要
		`self._stopping`	事件回圈完成的标志
	`call_soon`		呼叫该方法会立即将任务添加到待执行队列
	`run_once`		被`run_forever`呼叫，从`self._ready`队列里面取出任务执行
	`run_forever`		死回圈，若`self._stopping`则退出回圈
	`run_until_complete`		非常重要的函式，任务的起点和终点（后面详细介绍）
	`create_task`		将传入的函式封装成`task`物件，这个操作会将`task.__step`添加到`__ready`队列

`Handle`			所有的任务进入待执行队列（`Eventloop.call_soon`）之前都会封装成Handle物件
	`__init__`		初始化两个重要物件 `self._callback` 与 `self._args`
		`self._callback`	待执行函式主体
		`self._args`	待执行函式自变量
	`_run`		待执行函式执行

`get_event_loop`			获取当前执行绪的事件回圈

`_complete_eventloop`			将事件回圈的`_stopping`标志置位True

`run`			入口函式
`gather`			可以同时执行多个任务的入口函式	新增
`_GatheringFuture`			将每一个任务组成串列，封装成一个新的类	新增

tasks.py

类/函式	方法	物件	作用	描述
Task			继承自Future，主要用于整个协程运行的周期
	`__init__`		初始化物件 `self._coro` ，并且`call_soon`将`self.__step`加入`self._ready`队列
		`self._coro`	用户定义的函式主体
	`__step`		Task类的核心函式
	`__wakeup`		唤醒任务	新增

`ensure_future`			如果物件是一个Future物件，就回传，否则就会呼叫`create_task`回传，并且加入到`_ready`队列

futures.py

类/函式	方法	物件	作用	描述
Future			主要负责与用户函式进行互动
	`__init__`		初始化两个重要物件 `self._loop` 与 `self._callbacks`
		`self._loop`	事件回圈
		`self._callbacks`	回呼队列，任务暂存队列，等待时机成熟（状态不是`PENDING`），就会进入`_ready`队列
	`add_done_callback`		添加任务回呼函式，状态`_PENDING`，就虎进入`_callbacks`队列，否则进入`_ready`队列
	`set_result`		获取任务执行结果并存盘至`_result`，将状态置位`_FINISH`，呼叫`__schedule_callbacks`
	`__schedule_callbacks`		将回呼函式放入`_ready`，等待执行
	`result`		获取回传值
`__await__`			使用await就会进入这个方法	新增
`__iter__`			使用yield from就会进入这个方法	新增

3）执行程序

3.1）入口函式

main.py

    
if __name__ == "__main__":
    ret = wilsonasyncio.run(helloworld())
    print(ret)

ret = wilsonasyncio.run(helloworld())使用run，自变量是用户函式helloworld()，进入run，run的流程可以参考上一小节
run --> run_until_complete

3.2）事件回圈启动，同之前

3.3）第一次回圈run_forever --> run_once

将_ready队列的内容（即：task.__step）取出来执行，这里的coro是helloworld()

    def __step(self, exc=None):
        coro = self._coro
        try:
            if exc is None:
                result = coro.send(None)
            else:
                result = coro.throw(exc)
        except StopIteration as exc:
            super().set_result(exc.value)
        else:
            blocking = getattr(result, '_asyncio_future_blocking', None)
            if blocking:
                result._asyncio_future_blocking = False
                result.add_done_callback(self.__wakeup, result)
        finally:
            self = None

__step较之前的代码有改动
result = coro.send(None)，进入用户定义函式

async def helloworld():
    print('enter helloworld')
    ret = await wilsonasyncio.gather(hello(), world())
    print('exit helloworld')
    return ret

ret = await wilsonasyncio.gather(hello(), world())，这里没啥可说的，进入gather函式

def gather(*coros_or_futures, loop=None):
    loop = get_event_loop()

    def _done_callback(fut):
        nonlocal nfinished
        nfinished += 1

        if nfinished == nfuts:
            results = []
            for fut in children:
                res = fut.result()
                results.append(res)

            outer.set_result(results)

    children = []
    nfuts = 0
    nfinished = 0

    for arg in coros_or_futures:
        fut = tasks.ensure_future(arg, loop=loop)
        nfuts += 1
        fut.add_done_callback(_done_callback)

        children.append(fut)

    outer = _GatheringFuture(children, loop=loop)
    return outer

loop = get_event_loop()获取事件回圈
def _done_callback(fut)这个函式是回呼函式，细节后面分析，现在只需要知道任务（hello()与world()）执行完之后就会回呼就行
for arg in coros_or_futures for回圈确保每一个任务都是Future物件，并且add_done_callback将回呼函式设定为_done_callback ，还有将他们加入到_ready队列等待下一次回圈调度
3个重要的变量：
? ? ? ?children 里面存放的是每一个异步任务，在本例是hello()与world()
? ? ? ?nfuts 存放是异步任务的数量，在本例是2
? ? ? ?nfinished 存放的是异步任务完成的数量，目前是0，完成的时候是2
继续往下，来到了_GatheringFuture ，看看原始码：

class _GatheringFuture(Future):

    def __init__(self, children, *, loop=None):
        super().__init__(loop=loop)
        self._children = children

_GatheringFuture 最主要的作用就是将多个异步任务放入self._children，然后用_GatheringFuture 这个物件来管理，需要注意，这个物件继承了Future
至此，gather完成初始化，回传了outer，其实就是_GatheringFuture
总结一下gather，初始化了3个重要的变量，后面用来存放状态；给每一个异步任务添加回呼函式；将多个异步子任务合并，并且使用一个Future物件去管理

3.3.1）gather完成，回到helloworld()

async def helloworld():
    print('enter helloworld')
    ret = await wilsonasyncio.gather(hello(), world())
    print('exit helloworld')
    return ret

ret = await wilsonasyncio.gather(hello(), world()) gather回传_GatheringFuture ，随后使用await，就会进入Future.__await__

    def __await__(self):
        if self._state == _PENDING:
            self._asyncio_future_blocking = True
            yield self
        return self.result()

由于_GatheringFuture 的状态是_PENDING，所以进入if，遇到yield self，将self，也就是_GatheringFuture 回传（这里注意yield的用法，流程控制的功能）
那yield回到哪儿去了呢？从哪儿send就回到哪儿去，所以，他又回到了task.__step函式里面去

    def __step(self, exc=None):
        coro = self._coro
        try:
            if exc is None:
                result = coro.send(None)
            else:
                result = coro.throw(exc)
        except StopIteration as exc:
            super().set_result(exc.value)
        else:
            blocking = getattr(result, '_asyncio_future_blocking', None)
            if blocking:
                result._asyncio_future_blocking = False
                result.add_done_callback(self.__wakeup, result)
        finally:
            self = None

这里是本函式的第一个核心点，流程控制/跳转，需要非常的清晰，如果搞不清楚的同学，再详细的去阅读有关yield/yield from的文章
继续往下走，由于用户函式helloworld()没有结束，所以不会抛例外，所以来到了else分支
blocking = getattr(result, '_asyncio_future_blocking', None)这里有一个重要的状态，那就是_asyncio_future_blocking ，只有呼叫__await__，才会有这个自变量，默认是true，这个自变量主要的作用：一个异步函式，如果呼叫了多个子异步函式，那证明该异步函式没有结束（后面详细讲解），就需要添加“唤醒”回呼
result._asyncio_future_blocking = False将自变量置位False，并且添加self.__wakeup回呼等待唤醒
__step函式完成

这里需要详细讲解一下_asyncio_future_blocking 的作用

如果在异步函式里面出现了await，呼叫其他异步函式的情况，就会走到Future.__await__将_asyncio_future_blocking 设定为true

async def helloworld():
    print('enter helloworld')
    ret = await wilsonasyncio.gather(hello(), world())
    print('exit helloworld')
    return ret
    
class Future:
    def __await__(self):
        if self._state == _PENDING:
            self._asyncio_future_blocking = True
            yield self
        return self.result()

这样做了之后，在task.__step中就会把该任务的回呼函式设定为__wakeup
为啥要__wakeup，因为helloworld()并没有执行完成，所以需要再次__wakeup来唤醒helloworld()

这里揭示了，在Eventloop里面，只要使用await呼叫其他异步任务，就会挂起父任务，转而去执行子任务，直至子任务完成之后，回到父任务继续执行

先喝口水，休息一下，下面更复杂，，，

3.4）第二次回圈run_forever --> run_once

eventloops.py

    def run_once(self):
        ntodo = len(self._ready)
        for _ in range(ntodo):
            handle = self._ready.popleft()
            handle._run()

从队列中取出资料，此时_ready队列有两个任务，hello() world()，在gather的for回圈时添加的

async def hello():
    print('enter hello ...')
    return 'return hello ...'

async def world():
    print('enter world ...')
    return 'return world ...'

由于hello() world()没有await呼叫其他异步任务，所以他们的执行比较简单，分别一次task.__step就结束了，到达set_result()处
set_result()将回呼函式放入_ready队列，等待下次回圈执行

3.5）第三次回圈run_forever --> run_once

我们来看下回呼函式

    def _done_callback(fut):
        nonlocal nfinished
        nfinished += 1

        if nfinished == nfuts:
            results = []
            for fut in children:
                res = fut.result()
                results.append(res)

            outer.set_result(results)

没错，这是本文的第二个核心点，我们来仔细分析一下
这段代码最主要的逻辑，其实就是，只有当所有的子任务执行完之后，才会启动父任务的回呼函式，本文中只有hello() world()都执行完之后if nfinished == nfuts: ，才会启动父任务_GatheringFuture的回呼outer.set_result(results)
results.append(res)将子任务的结果取出来，放进父任务的results里面
子任务执行完成，终于到了唤醒父任务的时候了task.__wakeup

    def __wakeup(self, future):
        try:
            future.result()
        except Exception as exc:
            raise exc
        else:
            self.__step()
        self = None

3.6）第四次回圈run_forever --> run_once

future.result()从_GatheringFuture取出结果，然后进入task.__step

    def __step(self, exc=None):
        coro = self._coro
        try:
            if exc is None:
                result = coro.send(None)
            else:
                result = coro.throw(exc)
        except StopIteration as exc:
            super().set_result(exc.value)
        else:
            blocking = getattr(result, '_asyncio_future_blocking', None)
            if blocking:
                result._asyncio_future_blocking = False
                result.add_done_callback(self.__wakeup, result)
        finally:
            self = None

result = coro.send(None)其实就是helloworld() --> send又要跳回到当初yield的地方，那就是Future.__await__

    def __await__(self):
        if self._state == _PENDING:
            self._asyncio_future_blocking = True
            yield self
        return self.result()

return self.result()终于回传到helloworld()函式里面去了

async def helloworld():
    print('enter helloworld')
    ret = await wilsonasyncio.gather(hello(), world())
    print('exit helloworld')
    return ret

helloworld终于也执行完了，回传了ret

3.7）第五次回圈run_forever --> run_once

回圈结束
回到run

3.8）回到主函式，获取回传值

if __name__ == "__main__":
    ret = wilsonasyncio.run(helloworld())
    print(ret)

3.9）执行结果

? python3 main.py
enter helloworld
enter hello ...
enter world ...
exit helloworld
['return hello ...', 'return world ...']

● 终于结束了，这是一个非常长的小节了，但是我感觉很多细节还是没有说到，大家有问题请及时留言探讨
● _GatheringFuture一个非常重要的物件，它不但追踪了hello() world()的执行状态，唤醒helloworld()，并且将回传值传递给helloworld
● await async yield对流程的控制需要特别关注
● 本文中的代码，参考了python 3.7.7中asyncio的源代码，裁剪而来
● 本文中代码：代码