從頭造輪子：asyncio之 run_until_complete （1）

前言

今天開始聊一聊python3的asyncio。關於asyncio，大家肯定都有自己的理解，並且網上大神也把基礎概念也解釋的比較透徹。
本文寫作的初衷，主要是理解asyncio的原理並且實現一遍。
話不多說，我們開始！

一、知識準備

● 理解進程、線程、協程。簡單來說，這三個都是為了解決多任務同時進行的問題
  1）進程是操作資源分配的最小單位，多任務的實現主要是極快地在進程間來回切換，而進程切換消耗時間最長（系統調用）
  2）線程依賴於進程，多個線程共享了父進程的一部分資源，線程切換時間相對於進程來說消耗時間大大減少，但是由於python gil的存在，並不存在多線程（系統調用）
  3）協程依賴於線程，由於進程/線程切換都是系統調用，開銷是巨大的。而協程是在用戶空間內完成任務切換，不會切換到操作系統資源（寄存器、信號量、堆棧等），所以這種方式開銷最小。python的協程核心在於，遇到等待事件，就交出cpu控制權，轉而讓其他協程執行

● 理解python生成器，yield/yield from
  這裡就不班門弄斧了，直接推薦大佬的blog

● 理解關鍵字async/await，async/await是3.5之後的語法，和yield/yield from異曲同工

二、環境準備

組件 版本
python 3.7.7

三、run_until_complete的實現

先來看下官方asyncio的使用方法：

|># more main.py
import asyncio
async def hello():
    print('enter hello ...')
    return 'world'

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    task = loop.create_task(hello())
    rst = loop.run_until_complete(task)
    print(rst)
    
|># python3 main.py
enter hello ...
world

來看下造的輪子的使用方式：

▶ more main.py
from wilsonasyncio import get_event_loop
async def hello():
    print('enter hello ...')
    return 'world'


if __name__ == "__main__":
    loop = get_event_loop()
    task = loop.create_task(hello())
    rst = loop.run_until_complete(task)
    print(rst)
    
▶ python3 main.py
enter hello ...
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

tasks.py

類/函數 方法 對象 作用 描述
Task 繼承自Future，主要用於整個協程運行的周期
__init__ 初始化對象 self._coro ，並且call_soon將self.__step加入self._ready隊列
self._coro 用戶定義的函數主體
__step Task類的核心函數

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 獲取返回值

3）執行過程

3.1）入口函數

main.py

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

if __name__ == "__main__":
    loop = get_event_loop()
    task = loop.create_task(hello())
    rst = loop.run_until_complete(task)
    print(rst)
  • loop = get_event_loop()獲取事件循環
  • task = loop.create_task(hello())將用戶函數hello()封裝成協程，我們看下create_task的源碼
    def create_task(self, coro):
        task = tasks.Task(coro, loop=self)
        return task

    初始化一個Task對象，從代碼概覽得知，初始化對象之後會立即將__step添加到_ready隊列等待執行

  • rst = loop.run_until_complete(task)開始執行事件循環的第一個函數run_until_complete

3.2）事件循環啟動

eventloops.py

    def run_until_complete(self, future):
        future.add_done_callback(_complete_eventloop, future)
        self.run_forever()
        return future.result()
  • future.add_done_callback(_complete_eventloop, future)為future添加回調函數（future就是task對象，而task對象里的任務就是hello() ），回調函數是_complete_eventloop 。就是future執行完成之後執行_complete_eventloop
  • self.run_forever()啟動事件循環

3.3）第一次循環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隊列的內容（task.__step）取出來執行

tasks.py

    def __step(self, exc=None):
        coro = self._coro
        try:
            if exc is None:
                coro.send(None)
            else:
                coro.throw(exc)
        except StopIteration as exc:
            super().set_result(exc.value)
        finally:
            self = None
  • coro.send(None)核心代碼，跳轉回到用戶函數hello()

main.py

async def hello():
    print('enter hello ...')
    return 'world'
  • 用戶函數非常簡單，打印一行數據，以及返回一個字符串world，執行完成之後回到task.__step()
  • super().set_result(exc.value)由於用戶函數執行完成，會拋出異常StopIteration，捕獲之後執行set_result
  • 由代碼概覽得知set_result 的作用在於將任務狀態置位_FINISHED，並且將回調函數（_complete_eventloop ）寫入_ready隊列

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()
  • 繼續循環，handle封裝了_complete_eventloop
def _complete_eventloop(fut):
    fut._loop.stop()
  • 調用stop，設置停止標誌

3.5）第三次循環run_forever

    def run_forever(self):
        while True:
            self.run_once() 
            if self._stopping:
                break
  • 跳出事件循環，回到run_until_complete
    def run_until_complete(self, future):
        future.add_done_callback(_complete_eventloop, future)
        self.run_forever()
        return future.result()

3.6）回到主函數，獲取返回值

if __name__ == "__main__":
    loop = get_event_loop()
    task = loop.create_task(hello())
    rst = loop.run_until_complete(task)
    print(rst)
  • rst = loop.run_until_complete(task)獲取返回值

3.7）執行結果

▶ python3 main.py
enter hello ...
return world ...

五、流程總結

六、小結

● task對象與future有什麼區別？主要用於整個協程運行的周期，主要負責與用戶函數進行交互
● 本文從asyncio的第一個函數run_until_complete，介紹了asyncio的基本流程：用戶函數並不是立即執行，而是進入隊列，然後根據eventloop在合適的時機進行統一調度
● 本文中的代碼，參考了python 3.7.7中asyncio的源代碼，裁剪而來
● 本文中代碼：代碼

至此，本文結束
在下才疏學淺，有撒湯漏水的，請各位不吝賜教…

Tags:

VirMach 便宜 VPS

QNews

QNews