python異步編程之a(chǎn)syncio低階API的使用詳解
1.低階API介紹
asyncio中低階API的種類很多，涉及到開發(fā)的5個方面。包括：
- 獲取事件循環(huán)
- 事件循環(huán)方法集
- 傳輸
- 協(xié)議
- 事件循環(huán)策略
本篇中只講解asyncio常見常用的函數(shù)，很多底層函數(shù)如網(wǎng)絡、IPC、套接字、信號等不在本篇范圍。
2.獲取事件循環(huán)
事件循環(huán)是異步中重要的概念之一，用于驅動任務的執(zhí)行。包含的低階API如下：
| 函數(shù) | 功能 |
|---|---|
| asyncio.get_running_loop() | 獲取當前運行的事件循環(huán)首選函數(shù)。 |
| asyncio.get_event_loop() | 獲得一個事件循環(huán)實例 |
| asyncio.set_event_loop() | 將策略設置到事件循環(huán) |
| asyncio.new_event_loop() | 創(chuàng)建一個新的事件循環(huán) |
在asyncio初識這篇中提到過事件循環(huán)，可以把事件循環(huán)當做是一個while循環(huán)，在周期性的運行并執(zhí)行一些任務。這個說法比較抽象，事件循環(huán)本質上其實是能調用操作系統(tǒng)IO模型的模塊。以Linux系統(tǒng)為例，IO模型有阻塞，非阻塞，IO多路復用等。asyncio 常用的是IO多路復用模型的epool和 kqueue。事件循環(huán)原理涉及到異步編程的操作系統(tǒng)原理，后續(xù)更新一系列相關文章。
get_event_loop()創(chuàng)建一個事件循環(huán)，用于驅動協(xié)程的執(zhí)行
import asyncio
async def demo(i):
print(f"hello {i}")
def main():
loop = asyncio.get_event_loop()
print(loop._selector)
task = loop.create_task(demo(1))
loop.run_until_complete(task)
main()
結果：
<selectors.KqueueSelector object at 0x104eabe20>
hello 1
可以通過loop._selector屬性獲取到當前事件循環(huán)使用的是kqueue模型
獲取循環(huán)
import asyncio
async def demo(i):
res = asyncio.get_running_loop()
print(res)
print(f"hello {i}")
def main():
loop = asyncio.get_event_loop()
task = loop.create_task(demo(1))
loop.run_until_complete(task)
main()
結果：
<_UnixSelectorEventLoop running=True closed=False debug=False>
hello 1
推薦使用asyncio.run 創(chuàng)建事件循環(huán)，底層API主要用于庫的編寫。
3.生命周期
生命周期是用于管理任務的啟停的函數(shù)，如下：
| 函數(shù) | 功能 |
|---|---|
| loop.run_until_complete() | 運行一個期程/任務/可等待對象直到完成。 |
| loop.run_forever() | 一直運行事件循環(huán)，直到被顯示停止 |
| loop.stop() | 停止事件循環(huán) |
| loop.close() | 關閉事件循環(huán) |
| loop.is_running() | 返回 True ， 如果事件循環(huán)正在運行 |
| loop.is_closed() | 返回 True ，如果事件循環(huán)已經(jīng)被關閉 |
| await loop.shutdown_asyncgens() | 關閉異步生成器 |
run_until_complete：
運行一個期程/任務/可等待對象直到完成。run_until_complete的參數(shù)是一個futrue對象。當傳入一個協(xié)程，其內部會自動封裝成task。run_until_complete()是會自動關閉事件循環(huán)的函數(shù)，區(qū)別于run_forever()是需要手動關閉事件循環(huán)的函數(shù)。
import asyncio
async def demo(i):
print(f"hello {i}")
def main():
loop = asyncio.get_event_loop()
task = loop.create_task(demo(1))
# 傳入的是一個任務
loop.run_until_complete(task)
# 傳入的是一個協(xié)程也可以
loop.run_until_complete(demo(20))
main()
結果：
hello 1
hello 20
4.調試
| 函數(shù) | 功能 |
|---|---|
| loop.set_debug() | 開啟或禁用調試模式 |
| loop.get_debug() | 獲取當前測試模式 |
5.調度回調函數(shù)
在異步編程中回調函數(shù)是一種很常見的方法，想要在事件循環(huán)中增加一些回調函數(shù)，可以有如下方法：
| 函數(shù) | 功能 |
|---|---|
| loop.call_soon() | 盡快調用回調。 |
| loop.call_soon_threadsafe() | loop.call_soon() 方法線程安全的變體。 |
| loop.call_later() | 在給定時間之后調用回調函數(shù)。 |
| loop.call_at() | 在指定的時間調用回調函數(shù)。 |
這些回調函數(shù)既可以回調普通函數(shù)也可以回調協(xié)程函數(shù)。
call_soon函數(shù)原型：
loop.call_soon(callback, *args, context=None)
示例：
import asyncio
async def my_coroutine():
print("協(xié)程被執(zhí)行")
async def other_coro():
print("非call_soon調用")
def callback_function():
print("回調函數(shù)被執(zhí)行")
# 創(chuàng)建一個事件循環(huán)
loop = asyncio.get_event_loop()
# 使用create_task包裝協(xié)程函數(shù)，并調度執(zhí)行
loop.call_soon(loop.create_task, my_coroutine())
# 調度一個常規(guī)函數(shù)以盡快執(zhí)行
loop.call_soon(callback_function)
# 啟動一個事件循環(huán)
task = loop.create_task(other_coro())
loop.run_until_complete(task)
結果：
回調函數(shù)被執(zhí)行
非call_soon調用
協(xié)程被執(zhí)行
結果分析：
call_soon調用普通函數(shù)直接傳入函數(shù)名作為參數(shù)，調用協(xié)程函數(shù)需要講協(xié)程通過loop.create_task封裝成task。
6.線程/進程池
| 函數(shù) | 功能 |
|---|---|
| await loop.run_in_executor() | 多線程中運行一個阻塞的函數(shù) |
| loop.set_default_executor() | 設置 loop.run_in_executor() 默認執(zhí)行器 |
asyncio.run_in_executor 用于在異步事件循環(huán)中執(zhí)行一個阻塞的函數(shù)或方法。它將阻塞的調用委托給一個線程池或進程池，以確保不阻塞主事件循環(huán)?？梢杂糜谠趨f(xié)程中調用一些不支持異步編程的方法，不支持異步編程的模塊。
run_in_executor
import asyncio
import concurrent.futures
def blocking_function():
# 模擬一個阻塞的操作
import time
time.sleep(2)
return "阻塞函數(shù)返回"
async def async_function2():
print("async_function2 start")
await asyncio.sleep(1)
print("async_function2 end")
async def async_function():
print("異步函數(shù)開始執(zhí)行。。。")
print("調用同步阻塞函數(shù)")
# 使用run_in_executor調度執(zhí)行阻塞函數(shù)
result = await loop.run_in_executor(None, blocking_function)
print(f"獲取同步函數(shù)的結果: {result}")
# 創(chuàng)建一個事件循環(huán)
loop = asyncio.get_event_loop()
# 運行異步函數(shù)
loop.run_until_complete(asyncio.gather(async_function(), async_function2()))
結果：
異步函數(shù)開始執(zhí)行。。。
調用同步阻塞函數(shù)
async_function2 start
async_function2 end
獲取同步函數(shù)的結果: 阻塞函數(shù)返回
結果分析：
通過事件循環(huán)執(zhí)行任務async_function，在async_function中通過loop.run_in_executor調用同步阻塞函數(shù)blocking_function，該阻塞函數(shù)沒有影響事件循環(huán)中另一個任務async_function2的執(zhí)行。
await loop.run_in_executor(None, blocking_function)中None代表使用的是默認線程池，也可以替換成其他線程池。
使用自定義線程池和進程池
import asyncio
import concurrent.futures
def blocking_function():
# 模擬一個阻塞的操作
import time
time.sleep(2)
return "阻塞函數(shù)返回"
async def async_function():
print("異步函數(shù)開始執(zhí)行。。。")
print("調用同步阻塞函數(shù)")
# 線程池
with concurrent.futures.ThreadPoolExecutor() as pool:
result = await loop.run_in_executor(
pool, blocking_function)
print('線程池調用返回結果：', result)
# 進程池
with concurrent.futures.ProcessPoolExecutor() as pool:
result = await loop.run_in_executor(
pool, blocking_function)
print('進程池調用返回結果：', result)
if __name__ == '__main__':
# 創(chuàng)建一個事件循環(huán)
loop = asyncio.get_event_loop()
# 運行異步函數(shù)
loop.run_until_complete(async_function())
結果：
異步函數(shù)開始執(zhí)行。。。
調用同步阻塞函數(shù)
線程池調用返回結果： 阻塞函數(shù)返回
進程池調用返回結果： 阻塞函數(shù)返回
結果分析：
通過線程池concurrent.futures.ThreadPoolExecutor()和進程池concurrent.futures.ProcessPoolExecutor()執(zhí)行阻塞函數(shù)。
7.任務與期程
| 函數(shù) | 功能 |
|---|---|
| loop.create_future() | 創(chuàng)建一個 Future 對象。 |
| loop.create_task() | 將協(xié)程當作 Task 一樣調度。 |
| loop.set_task_factory() | 設置 loop.create_task() 使用的工廠，它將用來創(chuàng)建 Tasks 。 |
| loop.get_task_factory() | 獲取 loop.create_task() 使用的工廠，它用來創(chuàng)建 Tasks 。 |
create_futurecreate_future 的功能是創(chuàng)建一個future對象。future對象通常不需要手動創(chuàng)建，因為task會自動管理任務結果。相當于task是全自動，創(chuàng)建future是半自動。創(chuàng)建的future就需要手動的講future狀態(tài)設置成完成，才能表示task的狀態(tài)為完成。
import asyncio
def foo(future, result):
print(f"此時future的狀態(tài):{future}")
future.set_result(result)
print(f"此時future的狀態(tài):{future}")
if __name__ == '__main__':
loop = asyncio.get_event_loop()
# 手動創(chuàng)建future對象
all_done = loop.create_future()
# 設置一個回調函數(shù)用于修改設置future的結果
loop.call_soon(foo, all_done, "Future is done!")
result = loop.run_until_complete(all_done)
print("返回結果", result)
print("獲取future的結果", all_done.result())結果：
此時future的狀態(tài):<Future pending cb=[_run_until_complete_cb() at /Users/lib/python3.10/asyncio/base_events.py:184]>
此時future的狀態(tài):<Future finished result='Future is done!'>
返回結果 Future is done!
獲取future的結果 Future is done!
結果分析：
future設置結果之后之后，future對象的狀態(tài)就從pending變成finished狀態(tài)。如果一個future沒有手動設置結果，那么事件循環(huán)就不會停止。
create_task將協(xié)程封裝成一個task對象，事件循環(huán)主要操作的是task對象。協(xié)程沒有狀態(tài)，而task是有狀態(tài)的。
import asyncio
async def demo(i):
print(f"hello {i}")
await asyncio.sleep(1)
def main():
loop = asyncio.get_event_loop()
# 將攜程封裝成task，給事件使用
task = loop.create_task(demo(1))
loop.run_until_complete(task)
main()
>>>
hello 1
asyncio.create_task 和 loop.create_task的區(qū)別：
兩者實現(xiàn)的功能都是一樣的，將協(xié)程封裝成一個task，讓協(xié)程擁有了生命周期。區(qū)別僅僅在于使用的方法。asyncio.create_task 是高階API，不需要創(chuàng)建事件循環(huán)，而loop.create_task需要先創(chuàng)建事件循環(huán)再使用該方法。
到此這篇關于python異步編程之a(chǎn)syncio低階API的使用詳解的文章就介紹到這了,更多相關python asyncio低階API內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
Python3中條件控制、循環(huán)與函數(shù)的簡易教程
這篇文章主要給大家介紹了關于Python3中條件控制、循環(huán)與函數(shù)的相關資料，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧。2017-11-11
本文深入探討了 Django 中的請求與響應處理,從 Django 請求和響應的基礎知識、生命周期,到 HttpRequest 和 HttpResponse 對象的詳細介紹,同時,討論了 Django 的視圖和請求、響應處理,以及安全性和異步處理的考慮2023-09-09
python實現(xiàn)QQ定時發(fā)送新年祝福信息
大家好，本篇文章主要講的是python實現(xiàn)QQ定時發(fā)送新年祝福信息，感興趣的同學感快來看一看吧，對你有幫助的話記得收藏一下2022-02-02
這篇文章主要介紹了淺談TensorFlow之稀疏張量表示，具有很好的參考就價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-06-06
這篇文章主要介紹了如何使用ddddocr庫來識別滑塊驗證碼,并提供了一個示例代碼和識別結果,同時提醒注意ddddocr庫的大小限制,可能會影響某些無服務器函數(shù)的部署,需要的朋友可以參考下2024-11-11
這篇文章主要介紹了python urllib庫的使用詳解，幫助大家更好的利用python學習爬蟲，感興趣的朋友可以了解下2021-04-04
本文主要介紹了Flask wtforms實現(xiàn)表單驗證使用，文中通過示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2021-08-08
Python使用matplotlib實現(xiàn)在坐標系中畫一個矩形的方法
這篇文章主要介紹了Python使用matplotlib實現(xiàn)在坐標系中畫一個矩形的方法,涉及matplotlib模塊繪制圖形的相關技巧,需要的朋友可以參考下2015-05-05
Python+OpenCV 實現(xiàn)圖片無損旋轉90°且無黑邊
今天小編就為大家分享一篇Python+OpenCV 實現(xiàn)圖片無損旋轉90°且無黑邊，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-12-12
今天小編就為大家分享一篇通過pykafka接收Kafka消息隊列的方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2018-12-12