asyncio非同步模組的21個協程編寫實例

2020 年 8 月 25 日
筆記
Python

啟動一個無返回值協程

通過async關鍵字定義一個協程

import sys
import asyncio

async def coroutine():
    print('運行協程')
    
if sys.version_info >= (3, 7, 0):
    asyncio.run(coroutine())
else:
    loop = asyncio.get_event_loop()
    loop.run_until_complete(coroutine())
    loop.close()

輸出結果如下

運行協程

啟動一個有返回值協程

import sys
import asyncio

async def coroutine():
    print('運行協程')
    return 'done!'
  
if sys.version_info >= (3, 7, 0):
    result = asyncio.run(coroutine())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(coroutine())
    loop.close()
print(result)

輸出結果如下

運行協程
done!

使用await關鍵字

await關鍵字只能在一個協程內部使用

await等待另一個協程任務結果，它不會阻塞事件循環，但會在阻塞當前協程的上下文

import sys
import asyncio

async def coroutine():
    print('運行協程 調度asyncio.sleep協程任務')
    await asyncio.sleep(2)
    return 'done!'
  
print('啟動時間:', time.time())
if sys.version_info >= (3, 7, 0):
    result = asyncio.run(coroutine())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(coroutine())
    loop.close()
print(result)
print('結束時間:', time.time())

輸出結果如下

啟動時間: 1598236154.90588
運行協程 調度asyncio.sleep協程任務
done!
結束時間: 1598236156.9118028

回調函數: 迅速回調

迅速回調是立即主動調用一個方法

回調方法必然是一個普通函數

回調通過事件循環loop提供的方法添加，它不會阻塞協程上下文

同一個協程內的回調函數之間存在阻塞關係，不同協程內的回調函數之間無阻塞關係

import sys
import asyncio

def callback(a, c=2):
    time.sleep(c)
    print(f"傳入參數a={a}, c={c}, 時間{time.time()}")

async def main():
    print(f'添加回調, 時間{time.time()}')
    loop = asyncio.get_running_loop()
    loop.call_soon(callback, 1)
    loop.call_soon(callback, 2, 0)
    print(f'協程結束, 時間{time.time()}')

if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

輸出結果如下

添加回調, 時間1598236958.119624
協程結束, 時間1598236958.1196742
傳入參數a=1, c=2, 時間1598236960.1198878
傳入參數a=2, c=0, 時間1598236960.119946

回調函數:延時回調

延時回調可以指定在多久後執行回調函數

延時回調的其他特性和迅速回調一致

特別注意，如果事件循環內的協程已經運行結束，尚在等待調用的回調函數不會被調用
```
import sys
import asyncio

def callback(a, c=3):
    time.sleep(c)
    print(f"傳入參數a={a}, c={c}, 時間{time.time()}")

async def main():
    print(f'添加回調, 時間{time.time()}')
    loop = asyncio.get_running_loop()
    loop.call_later(1, callback, 2, 0)
    loop.call_soon(callback, 1)

    print(f'協程結束, 時間{time.time()}')

if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()
```
輸出結果如下
```
添加回調, 時間1598237313.375363
協程結束, 時間1598237313.375415
傳入參數a=1, c=3, 時間1598237316.376031
傳入參數a=2, c=0, 時間1598237316.376389
```
解釋說明，對於協程main，call_later和call_soon是不會阻塞協程上下文的，因此兩個回調函數可以視為同時加入了事件循環的回調，其中回調callback(1)方法是立即調用，回調函數會sleep 3秒，而回調方法callback(2, 0)是在協程延時 1秒後調用，但回調函數之間是存在阻塞關係的，因此它會等待callback(1)先運行結束，然後判斷是否滿足了延時條件再執行。

回調函數:定時回調

此處的指定時間並非系統時間戳，而是指事件循環創建的時間戳

通過loop.time()獲取事件循環時間戳

import sys
import asyncio

def callback(a, loop, c=3):
    time.sleep(c)
    print(f"傳入參數a={a}, c={c}, 時間{loop.time()}")


async def main():
    loop = asyncio.get_running_loop()
    now = loop.time()
    print(f'事件循環時間戳{now}')
    loop.call_at(now + 4, callback, 2, loop, 1)
    loop.call_soon(callback, 1, loop)
    await asyncio.sleep(5)
    print(f'協程結束, 時間戳{loop.time()}')

if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

asyncio.Future對象

Future的實例化對象可以認為是一個協程對象，它可以使用await關鍵字

通過延時回調+Future可以模擬一個協程的邏輯

即延時回調 ≈ 非同步等待返回結果

Future ≈ 非阻塞模型，它不會阻塞事件循環的其他協程

import sys
import asyncio

def mark_done(future: asyncio.Future, result):
    print(f'標記future結束  時間{time.time()}')
    future.set_result(result)


async def main():
    print(f'協程開始  時間{time.time()}')
    loop = asyncio.get_running_loop()
    future = asyncio.Future()
    loop.call_later(3, mark_done, future, 'done!')
    result = await future
    print(f'future結果 {result}  時間{time.time()}')


if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

輸出結果如下

協程開始  時間1598238858.6768472
標記future結束  時間1598238861.678293
future結果 done!  時間1598238861.678438

Future回調

Future的回調是在一個Future對象標記結束後運行的函數

await會等待回調函數執行結束，即回調函數會在協程上下文阻塞

import sys
import asyncio

def callback(future: asyncio.Future):
    # 基於Future的回調函數一定要接受一個Future對象
    print(f'Future回調被調用 時間{time.time()}')
    time.sleep(1)


async def main():
    print(f'協程開始  時間{time.time()}')
    loop = asyncio.get_running_loop()
    future = asyncio.Future()
    future.add_done_callback(callback)
    loop.call_later(1, lambda future: future.set_result('done'), future)
    result = await future
    print(f'future結果 {result} 時間{time.time()}')


if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

輸出結果如下

協程開始  時間1598241100.534262
Future回調被調用 時間1598241101.539032
future結果 done 時間1598241102.5411909

aysncio.Task對象

Task對象在調用create_task會立即執行，它類似迅速回調

asyncio.create_task是python3.7引入的高級api，3.7以下使用ensure_future方法

與回調函數的區別:

回調函數一定不是協程，而Task對象只能創建協程任務
回調函數的傳入參數直接通過添加回調的方法傳入，而任務對象直接傳入協程形參
回調函數之間運行是阻塞的，而Task對象則是基於事件循環的標準協程

在協程上下文中可以使用await關鍵字等待任務結果

import sys
import asyncio

async def task_func(n):
    print(f'運行 task_func  時間{time.time()}')
    await asyncio.sleep(n)
    return 'task done!'


async def main():
    loop = asyncio.get_running_loop()
    task = loop.create_task(task_func(3))
    result = await task
    print(f'task結果 {result} 時間{time.time()}')


if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

輸出結果如下

運行 task_func  時間1598241581.334356
task結果 task done! 時間1598241584.338337

Task取消任務

可以取消一個正在事件循環內運行的task對象

Task和Future一樣，支援通過add_done_callback添加回調函數

import sys
import asyncio

async def task_func(n):
    print(f'運行 task_func  時間{time.time()}')
    await asyncio.sleep(n)
    return 'task done!'


async def main():
    loop = asyncio.get_running_loop()
    task = asyncio.create_task(task_func(5))
    loop.call_later(3, lambda task: task.cancel(), task)
    # Task被取消 會拋出CancelledError異常
    try:
        result = await task
        print(f'task結果 {result} 時間{time.time()}')
    except asyncio.exceptions.CancelledError:
        print(f'task被取消了 時間{time.time()}')


if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

輸出結果如下

運行 task_func  時間1598242256.3299708
task被取消了 時間1598242259.335436

使用asyncio.ensure_future創建任務

asyncio.ensure_future實現效果和asyncio.create_task一致

import sys
import asyncio

async def task_func(n):
    print(f'運行 task_func  時間{time.time()}')
    await asyncio.sleep(n)
    return 'task done!'


async def main():
    print(f'協程開始  時間{time.time()}')
    task = asyncio.ensure_future(task_func(2))
    result = await task
    print(f'task結果 {result} 時間{time.time()}')


if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

輸出結果如下

協程開始  時間1598249260.1346428
運行 task_func  時間1598249260.1347158
task結果 task done! 時間1598249262.137878

asyncio.wait執行多個協程

asyncio.wait接受一個協程列表/元組將其加入事件循環

返回兩個列表，分別包含已完成和正在執行的Future對象

asyncio.wait 會阻塞協程上下文直至滿足指定的條件（默認條件所有協程運行結束）

wait支援設置一個超時時間，但在超時發生時不會取消可等待對象，但若事件循環結束時未完成則會拋出CancelledError異常。如果要超時主動取消，可用wait_for方法

asyncio.wait 返回的結果集是按照事件循環中的任務完成順序排列的，所以通常和原始任務順序不同

import sys
import asyncio

async def corn_sleep(n):
    try:
        await asyncio.sleep(n)
    except asyncio.exceptions.CancelledError:
        print(f'corn({n})超時取消!  時間{time.time()}')
    print(f'corn({n}) done!  時間{time.time()}')
    return f'corn({n}) done!  時間{time.time()}'


async def main():
    print(f'協程開始  時間{time.time()}')
    tasks = [corn_sleep(i) for i in range(1, 6)]
    done, pending = await asyncio.wait(tasks, timeout=3)
    for task in done:
        pass
        # print(task.result())
    for task in pending:
        print(task.done())
    await asyncio.sleep(1)
    print(f'協程結束  時間{time.time()}')


if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

輸出結果如下

協程開始  時間1598253388.614105
corn(1) done!  時間1598253389.6196852
corn(2) done!  時間1598253390.6171541
False
False
False
corn(3) done!  時間1598253391.617078
corn(4) done!  時間1598253392.617147
協程結束  時間1598253392.617195
corn(5)超時取消!  時間1598253392.6173718
corn(5) done!  時間1598253392.6173818

asyncio.gather執行多個協程

gather方法和wait都可以執行多個協程，但輸入和輸出有所差異

輸出差異，gather保證了結果列表的順序，它是嚴格遵循傳入任務順序的
輸入差異，wait方法接受的是一個協程列表，而gather是通過可變長參數傳入協程方法的
wait支援超時設置，gather無法設置超時時間
wait返回兩個列表，列表元素是Future對象，gather只返回done列表，列表元素是Future對象的result()結果

import sys
import asyncio

async def corn_sleep(n):
    await asyncio.sleep(n)
    return f'corn({n}) done!  時間{time.time()}'


async def main():
    print(f'協程開始  時間{time.time()}')
    tasks = [corn_sleep(i) for i in range(1, 6)]
    done = await asyncio.gather(*tasks)
    for result in done:
        print(result)
    print(f'協程結束  時間{time.time()}')


if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

輸出結果如下

協程開始  時間1598250547.5619462
corn(1) done!  時間1598250548.567125
corn(2) done!  時間1598250549.5660028
corn(3) done!  時間1598250550.563731
corn(4) done!  時間1598250551.566098
corn(5) done!  時間1598250552.566897
協程結束  時間1598250552.566983

asyncio.as_completed執行多個協程

as_completed方法功能同wait和gather都可以用於執行多個協程

as_completed接受的是協程列表，返回的是一個迭代器，迭代元素為Future對象

as_completed方法支援超時設置，但它會在協程上下文拋出asyncio.exceptions.TimeoutError錯誤

as_completed方法返回結果集是無序的

import sys
import asyncio

async def corn_sleep(n):
    await asyncio.sleep(n)
    return f'corn({n}) done!  時間{time.time()}'


async def main():
    print(f'協程開始  時間{time.time()}')
    tasks = [corn_sleep(i) for i in range(1, 6)]
    for task in asyncio.as_completed(tasks, timeout=3.5):
        try:
            result = await task
            print(result)
        except asyncio.exceptions.TimeoutError:
            print(f'協程超時了  時間{time.time()}')
            break
    print(f'協程結束  時間{time.time()}')


if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

輸出結果如下

協程開始  時間1598251179.489662
corn(1) done!  時間1598251180.493866
corn(2) done!  時間1598251181.4911182
corn(3) done!  時間1598251182.493823
協程超時了  時間1598251182.991565
協程結束  時間1598251182.991592

協程鎖

相比執行緒、進程鎖，協程鎖似乎應用場景不那麼多，比如保存文件時?

import sys
import asyncio
from functools import partial

def callback(lock: asyncio.Lock):
    print(f'釋放鎖  時間{time.time()}')
    lock.release()


async def corn1(x, loop, lock):
    async with lock:
        pass
    # 加鎖，利用延遲回調1秒後解鎖
    await lock.acquire()
    loop.call_later(1, callback, lock)
    return f'{x} done! 時間{time.time()}'


async def main():
    loop = asyncio.get_running_loop()
    lock = asyncio.Lock()
    await lock.acquire()
    print(f'加鎖 時間{time.time()}')
    loop.call_later(2, callback, lock)
    corn = partial(corn1, loop=loop, lock=lock)
    results = await asyncio.gather(corn(1), corn(2), corn(3))
    for result in results:
        print(result)


if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

輸出結果如下

加鎖 時間1598251855.600318
釋放鎖  時間1598251857.6041799
釋放鎖  時間1598251858.606617
釋放鎖  時間1598251859.609798
1 done! 時間1598251857.604349
2 done! 時間1598251858.606834
3 done! 時間1598251859.6099358

Event 事件對象

Event對象和Lock對象有一定的相似性，可用於同步操作

Event對象提供一個狀態標記位，它是一個布爾值，只用來判斷狀態

import sys
import asyncio

def task_done(event: asyncio.Event):
    event.set()


async def corn(event: asyncio.Event):
    await event.wait()
    print(f'corn done!  時間{time.time()}')


async def main():
    print(f'協程開始  時間{time.time()}')
    event = asyncio.Event()
    loop = asyncio.get_running_loop()
    loop.call_later(2, task_done, event)
    await corn(event)


if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

輸出結果如下

協程開始  時間1598252539.2691798
corn done!  時間1598252541.273142

Queue 協程隊列

與普通隊列相比，協程隊列有一個task_done方法，用於標記某一個任務結束，而join方法則會阻塞，直到滿足從隊列取出的個數等於task_done方法調用的次數

import sys
import asyncio

async def producer(queue: asyncio.Queue):
    print('生產者 上班')
    for i in range(1, 6):
        await queue.put(f'產品({i})')
        await asyncio.sleep(2)
    await queue.put(None)
    await queue.join()
    print('生產者 打烊')


async def consumer(x, queue: asyncio.Queue):
    print(f'消費者{x}號進場')
    while True:
        pt = await queue.get()
        queue.task_done()
        if pt is None:
            await queue.put(None)
            break
        else:
            print(f'{x}號 消費了 {pt}')


async def main():
    loop = asyncio.get_running_loop()
    queue = asyncio.Queue(maxsize=3)
    loop.create_task(producer(queue))
    await asyncio.wait([consumer(i, queue) for i in range(3)])

if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

輸出結果如下

生產者 上班
消費者2號進場
2號 消費了 產品(1)
消費者0號進場
消費者1號進場
2號 消費了 產品(2)
0號 消費了 產品(3)
1號 消費了 產品(4)
2號 消費了 產品(5)
生產者 下班

asyncio.subprocess 非同步調用子進程

asyncio提供create_subprocess_exec和create_subprocess_shell方法

前者可以調用任意程式，後者則通過shell命令行調用其他程式

import sys
import asyncio

async def get_date():
    code = 'import datetime; print(datetime.datetime.now())'
    task1 = asyncio.create_subprocess_exec(
        sys.executable, '-c', code,
        stdout=asyncio.subprocess.PIPE
    )
    task2 = asyncio.create_subprocess_shell(
        'ls -all ~/Desktop', stdout=asyncio.subprocess.PIPE
    )
    results = await asyncio.gather(task1, task2)
    response = []
    for proc in results:
        data = await proc.stdout.read()
        line = data.decode('utf8').rstrip()
        response.append(line)
    return '\n'.join(response)

date = asyncio.run(get_date())
print(f"當前時間: {date}")

輸出結果如下

當前時間: 2020-08-24 16:11:18.143943
total 16
drwxr-xr-x   3 sw  staff    96 Apr  6 09:11 $RECYCLE.BIN
drwx------@  8 sw  staff   256 Jun 18 10:52 .
drwxr-xr-x+ 63 sw  staff  2016 Aug 23 20:52 ..
-rw-r--r--@  1 sw  staff  6148 Jul 22 17:54 .DS_Store
-rw-r--r--   1 sw  staff     0 May 27  2019 .localized
drwxr-xr-x   3 sw  staff    96 Feb  3  2020 Don't Starve Together.app
drwxr-xr-x   9 sw  staff   288 May  3 18:24 HeavenMS
drwxr-xr-x   3 sw  staff    96 Sep 20  2019 Tomb Raider.app

網路通訊-高級API

asyncio封裝了幾個高級方法來快速實現網路通訊

await asyncio.open_connection() 建立TCP連接
await asyncio.open_unix_connection() 建立Unix socket連接
await start_server() 啟動TCP服務
await start_unix_server() 啟動Unix socket服務
StreamReader 接收網路數據的高級async/await對象
StreamWriter 發送網路數據的高級async/await對象

以搭建一個簡單http伺服器為例

import sys
import asyncio

async def http_handle(render: asyncio.StreamReader,
                      writer: asyncio.WriteTransport):
    """在asyncio.start_server傳入該協程
    當一個tcp連接建立時就會調用該回調，並傳入兩個參數
    :param render: StreamReader對象
    :param writer: StreamWriter對象
    StreamReader和StreamWriter對象都是繼承於Transport
    :return:
    """
    message = await render.read(1024)
    print(message.decode())
    writer.write(b'HTTP/1.1 200 OK\r\n\r\nHello Client!')
    writer.close()


async def main():
    server = await asyncio.start_server(http_handle, '127.0.0.1', 8000)
    addr = server.sockets[0].getsockname()
    print(f'伺服器啟動 {addr}')
    async with server:
        await server.serve_forever()


if sys.version_info >= (3, 7, 0):
    result = asyncio.run(main())
else:
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    loop.close()

瀏覽器打開//127.0.0.1:8000 控制台輸出結果如下

GET / HTTP/1.1
Host: 127.0.0.1:8000
Connection: keep-alive
Cache-Control: max-age=0
Upgrade-Insecure-Requests: 1
User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.135 Safari/537.36
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9
Sec-Fetch-Site: none
Sec-Fetch-Mode: navigate
Sec-Fetch-User: ?1
Sec-Fetch-Dest: document
Accept-Encoding: gzip, deflate, br
Accept-Language: zh-CN,zh;q=0.9

瀏覽器顯示結果

Hello Client!

網路通訊-低級API

案例19所用的方法是基於底層Transport和Protocol封裝的結果

Protocol協議類是通過重寫一系列回調方法來處理網路請求

Transport實例可以簡單看作是對一個socket對象封裝，實現了數據收發功能

除了網路通訊外，子進程（進程通訊）也是通過協議回調來實現

以一個tcp服務端為例，較常用的有4個回調函數

connection_made() 連接建立時被調用
connection_lost() 連接丟失或關閉時被調用
data_received() 接收到數據時被調用
eof_received() 接收到EOF時被調用

以案例19的http服務端為例，實現相同效果

import sys
import asyncio

class HttpProtocol(asyncio.Protocol):
    def __init__(self):
        """每個tcp連接建立時都會實例化這個類"""
        self.transport: asyncio.Transport = None

    def connection_made(self, transport: asyncio.Transport):
        """連接建立時被調用"""
        self.transport = transport

    def data_received(self, data):
        """接收數據時被調用"""
        print(data.decode())
        self.transport.write(b'HTTP/1.1 200 OK\r\n\r\nHello Client!')
        self.transport.close()


loop = asyncio.get_event_loop()
server = loop.create_server(HttpProtocol, '127.0.0.1', 8000)
loop.run_until_complete(server)
loop.run_forever()

訊號處理

事件循環能夠支援添加訊號來執行回調函數

import signal
import asyncio

def ask_exit(sig_name: str, loop: asyncio.AbstractEventLoop):
    print("捕獲訊號 %s: exit" % sig_name)
    loop.stop()


async def main():
    loop = asyncio.get_running_loop()
    for signame in {'SIGINT', 'SIGTERM'}:
        loop.add_signal_handler(
            getattr(signal, signame),
            partial(ask_exit, signame, loop))

    await asyncio.sleep(3600)

print("事件循環將在1小時後或者按下Ctrl+C停止")
try:
    asyncio.run(main())
except RuntimeError as e:
    if str(e) != 'Event loop stopped before Future completed.':
        raise

輸出結果如下

事件循環將在1小時後或者按下Ctrl+C停止
^C捕獲訊號 SIGINT: exit

高頻方法總結

回調和Task具有一定相似性，回調是函數，任務是協程

向事件循環非阻塞運行一個協程是創建一個任務，可以使用如下方法

方法	傳入	返回	備註
asyncio.create_task	協程	Task對象	高層級api、Python3.7 +
asyncio.ensure_future	協程	Task對象	高層級api
loop.create_task	協程	Task對象	低層級api

向事件循環添加多個協程（使用await關鍵詞才會實際添加運行），可使用如下方法

方法	傳入	返回	備註
asyncio.wait	協程列表	done/pending列表元素是Future對象	支援超時設置，結果無序
asyncio.gather	變長參數	協程返回值組成的列表	返回結果列表有序
asyncio.as_completed	協程列表	協程任務迭代器	通過for循環遍歷獲取任務結果

在unix平台，可以使用性能更好uvloop來替代asyncio默認的事件循環
```
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
```

Tags: Python