前言

書接上文，本文造第三個(gè)輪子，也是asyncio包里面非常常用的一個(gè)函數(shù)gather

創(chuàng)新互聯(lián)是專業(yè)的北海網(wǎng)站建設(shè)公司，北海接單;提供網(wǎng)站設(shè)計(jì)制作、成都做網(wǎng)站,網(wǎng)頁設(shè)計(jì),網(wǎng)站設(shè)計(jì),建網(wǎng)站,PHP網(wǎng)站建設(shè)等專業(yè)做網(wǎng)站服務(wù);采用PHP框架,可快速的進(jìn)行北海網(wǎng)站開發(fā)網(wǎng)頁制作和功能擴(kuò)展;專業(yè)做搜索引擎喜愛的網(wǎng)站,專業(yè)的做網(wǎng)站團(tuán)隊(duì),希望更多企業(yè)前來合作!

一、知識(shí)準(zhǔn)備

● 相對(duì)于前兩個(gè)函數(shù)，gather的使用頻率更高，因?yàn)樗С侄鄠€(gè)協(xié)程任務(wù)“同時(shí)”執(zhí)行
● 理解__await__ __iter__的使用
● 理解關(guān)鍵字async/await，async/await是3.5之后的語法，和yield/yield from異曲同工
● 今天的文章有點(diǎn)長(zhǎng)，請(qǐng)大家耐心看完

二、環(huán)境準(zhǔn)備

三、gather的實(shí)現(xiàn)

先來看下官方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 ...']

自己造的輪子也很好的運(yùn)行了，下面我們來看下輪子的代碼

四、代碼解析

輪子代碼

1）代碼組成

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

2）代碼概覽：

eventloops.py

tasks.py

futures.py

3）執(zhí)行過程

3.1）入口函數(shù)

main.py

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

• ret = wilsonasyncio.run(helloworld())使用run，參數(shù)是用戶函數(shù)helloworld()，進(jìn)入run，run的流程可以參考上一小節(jié)
• run --> run_until_complete

3.2）事件循環(huán)啟動(dòng)，同之前

3.3）第一次循環(huán)run_forever --> run_once

• 將_ready隊(duì)列的內(nèi)容（即：task.__step）取出來執(zhí)行，這里的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較之前的代碼有改動(dòng)
• result = coro.send(None)，進(jìn)入用戶定義函數(shù)

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

• ret = await wilsonasyncio.gather(hello(), world())，這里沒啥可說的，進(jìn)入gather函數(shù)

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()獲取事件循環(huán)
• def _done_callback(fut)這個(gè)函數(shù)是回調(diào)函數(shù)，細(xì)節(jié)后面分析，現(xiàn)在只需要知道任務(wù)（hello()與world()）執(zhí)行完之后就會(huì)回調(diào)就行
• for arg in coros_or_futures for循環(huán)確保每一個(gè)任務(wù)都是Future對(duì)象，并且add_done_callback將回調(diào)函數(shù)設(shè)置為_done_callback，還有將他們加入到_ready隊(duì)列等待下一次循環(huán)調(diào)度
• 3個(gè)重要的變量：
  ? ? ? ?children里面存放的是每一個(gè)異步任務(wù)，在本例是hello()與world()
  ? ? ? ?nfuts存放是異步任務(wù)的數(shù)量，在本例是2
  ? ? ? ?nfinished存放的是異步任務(wù)完成的數(shù)量，目前是0，完成的時(shí)候是2
• 繼續(xù)往下，來到了_GatheringFuture，看看源碼：

class _GatheringFuture(Future):

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

• _GatheringFuture最主要的作用就是將多個(gè)異步任務(wù)放入self._children，然后用_GatheringFuture這個(gè)對(duì)象來管理。需要注意，這個(gè)對(duì)象繼承了Future
• 至此，gather完成初始化，返回了outer，其實(shí)就是_GatheringFuture
• 總結(jié)一下gather，初始化了3個(gè)重要的變量，后面用來存放狀態(tài)；給每一個(gè)異步任務(wù)添加回調(diào)函數(shù)；將多個(gè)異步子任務(wù)合并，并且使用一個(gè)Future對(duì)象去管理

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，就會(huì)進(jìn)入Future.__await__

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

• 由于_GatheringFuture的狀態(tài)是_PENDING，所以進(jìn)入if，遇到yield self，將self，也就是_GatheringFuture返回（這里注意yield的用法，流程控制的功能）
• 那yield回到哪兒去了呢？從哪兒send就回到哪兒去，所以，他又回到了task.__step函數(shù)里面去

    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

• 這里是本函數(shù)的第一個(gè)核心點(diǎn)，流程控制/跳轉(zhuǎn)，需要非常的清晰，如果搞不清楚的同學(xué)，再詳細(xì)的去閱讀有關(guān)yield/yield from的文章
• 繼續(xù)往下走，由于用戶函數(shù)helloworld()沒有結(jié)束，所以不會(huì)拋異常，所以來到了else分支
• blocking = getattr(result, '_asyncio_future_blocking', None)這里有一個(gè)重要的狀態(tài)，那就是_asyncio_future_blocking，只有調(diào)用__await__，才會(huì)有這個(gè)參數(shù)，默認(rèn)是true，這個(gè)參數(shù)主要的作用：一個(gè)異步函數(shù)，如果調(diào)用了多個(gè)子異步函數(shù)，那證明該異步函數(shù)沒有結(jié)束（后面詳細(xì)講解），就需要添加“喚醒”回調(diào)
• result._asyncio_future_blocking = False將參數(shù)置位False，并且添加self.__wakeup回調(diào)等待喚醒
• __step函數(shù)完成

這里需要詳細(xì)講解一下_asyncio_future_blocking的作用

• 如果在異步函數(shù)里面出現(xiàn)了await，調(diào)用其他異步函數(shù)的情況，就會(huì)走到Future.__await__將_asyncio_future_blocking設(shè)置為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中就會(huì)把該任務(wù)的回調(diào)函數(shù)設(shè)置為__wakeup
• 為啥要__wakeup，因?yàn)?code>helloworld()并沒有執(zhí)行完成，所以需要再次__wakeup來喚醒helloworld()

這里揭示了，在Eventloop里面，只要使用await調(diào)用其他異步任務(wù)，就會(huì)掛起父任務(wù)，轉(zhuǎn)而去執(zhí)行子任務(wù)，直至子任務(wù)完成之后，回到父任務(wù)繼續(xù)執(zhí)行

先喝口水，休息一下，下面更復(fù)雜。。。

3.4）第二次循環(huán)run_forever --> run_once

eventloops.py

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

• 從隊(duì)列中取出數(shù)據(jù)，此時(shí)_ready隊(duì)列有兩個(gè)任務(wù)，hello() world()，在gather的for循環(huán)時(shí)添加的

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

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

• 由于hello() world()沒有await調(diào)用其他異步任務(wù)，所以他們的執(zhí)行比較簡(jiǎn)單，分別一次task.__step就結(jié)束了，到達(dá)set_result()處
• set_result()將回調(diào)函數(shù)放入_ready隊(duì)列，等待下次循環(huán)執(zhí)行

3.5）第三次循環(huán)run_forever --> run_once

• 我們來看下回調(diào)函數(shù)

    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)

• 沒錯(cuò)，這是本文的第二個(gè)核心點(diǎn)，我們來仔細(xì)分析一下
• 這段代碼最主要的邏輯，其實(shí)就是，只有當(dāng)所有的子任務(wù)執(zhí)行完之后，才會(huì)啟動(dòng)父任務(wù)的回調(diào)函數(shù)，本文中只有hello() world()都執(zhí)行完之后if nfinished == nfuts: ，才會(huì)啟動(dòng)父任務(wù)_GatheringFuture的回調(diào)outer.set_result(results)
• results.append(res)將子任務(wù)的結(jié)果取出來，放進(jìn)父任務(wù)的results里面
• 子任務(wù)執(zhí)行完成，終于到了喚醒父任務(wù)的時(shí)候了task.__wakeup

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

3.6）第四次循環(huán)run_forever --> run_once

• future.result()從_GatheringFuture取出結(jié)果，然后進(jìn)入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)其實(shí)就是helloworld() --> send又要跳回到當(dāng)初yield的地方，那就是Future.__await__

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

• return self.result()終于返回到helloworld()函數(shù)里面去了

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

• helloworld終于也執(zhí)行完了，返回了ret

3.7）第五次循環(huán)run_forever --> run_once

• 循環(huán)結(jié)束
• 回到run

3.8）回到主函數(shù)，獲取返回值

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

3.9）執(zhí)行結(jié)果

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

五、流程總結(jié)

六、小結(jié)

● 終于結(jié)束了，這是一個(gè)非常長(zhǎng)的小節(jié)了，但是我感覺很多細(xì)節(jié)還是沒有說到，大家有問題請(qǐng)及時(shí)留言探討
● _GatheringFuture一個(gè)非常重要的對(duì)象，它不但追蹤了hello() world()的執(zhí)行狀態(tài)，喚醒helloworld()，并且將返回值傳遞給helloworld
● await async yield對(duì)流程的控制需要特別關(guān)注
● 本文中的代碼，參考了python 3.7.7中asyncio的源代碼，裁剪而來
● 本文中代碼：代碼

至此，本文結(jié)束
在下才疏學(xué)淺，有撒湯漏水的，請(qǐng)各位不吝賜教...
更多文章，請(qǐng)關(guān)注我：wilson.chai