前提概要：python因為GIL鎖，所以運行都是單線程，導致python運行的速度慢，為此要解決這個問題有多進程、多線程，但是使用這些方法，我們就要多加考慮線程安全問題，顧很麻煩，所以推出了協程。協程運行在線程上，所以一樣是單線程，但是卻能實現并發，遇見io耗時操作a，會把這個a操作掛后臺執行，程序接著執行下一個操作b，當后臺操作a結束后，程序再回去解決操作a的返回結果。類似前端js的編程思想。當然了解決這個python并發問題還有celery等等很多方法。

寫這篇的目的：因為曾經寫爬蟲使用過asyncio，后來后端變成使用go，所以最近感覺有點遺忘這個語法，因此寫下一篇記錄一下（參考文檔：asyncio --- 異步 I/O — Python 3.10.2 文檔）

一、基本使用

（1）基礎語法

能使用await必須是可等待對象：協程、任務、Future.

協程：帶有async的函數是協程函數，可等待

任務：用asyncio.create_task()方法創建的任務

Future：是一種特殊的?低層級?可等待對象，表示一個異步操作的?最終結果。當一個 Future 對象?被等待，這意味著協程將保持等待直到該 Future 對象在其他地方操作完畢。在 asyncio 中需要 Future 對象以便允許通過 async/await 使用基于回調的代碼。通常情況下?沒有必要?在應用層級的代碼中創建 Future 對象。Future 對象有時會由庫和某些 asyncio API 暴露給用戶，用作可等待對象。

import asyncio import timeasync def say_after(delay, what):await asyncio.sleep(delay)print(what)async def main():# 創建任務 task1、task2task1 = asyncio.create_task(say_after(1, 'hello'))task2 = asyncio.create_task(say_after(2, 'world'))print(f"started at {time.strftime('%X')}")# 執行這兩個任務await task1await task2print(f"finished at {time.strftime('%X')}")# 不能普通的調用main(),否則就不是協程操作 asyncio.run(main())""" 運行結果：started at 09:36:34 hello world finished at 09:36:36"""

（2）任務執行的結果，存入并返回一個列表

import asyncioasync def factorial(name, number):f = 1for i in range(2, number + 1):print(f"Task {name}: Compute factorial({number}), currently i={i}...")await asyncio.sleep(1)f *= iprint(f"Task {name}: factorial({number}) = {f}")return fasync def main():# 異步執行，這3個任務:L = await asyncio.gather(factorial("A", 2),factorial("B", 3),factorial("C", 4),)print(L)asyncio.run(main())"""執行結果：放在一塊的，是并發執行出來的結果Task A: Compute factorial(2), currently i=2... Task B: Compute factorial(3), currently i=2... Task C: Compute factorial(4), currently i=2...Task A: factorial(2) = 2 Task B: Compute factorial(3), currently i=3... Task C: Compute factorial(4), currently i=3...Task B: factorial(3) = 6 Task C: Compute factorial(4), currently i=4...Task C: factorial(4) = 24 [2, 6, 24]"""

（3）超時：

? ? ? ? （3.1）wait_for()--->超時會被取消

import asyncioasync def eternity():# 休眠1個小時await asyncio.sleep(3600)print('yay!')async def main():# 超時需要使用異常來抓取try:await asyncio.wait_for(eternity(), timeout=1.0)except asyncio.TimeoutError:print('timeout!')asyncio.run(main())# 執行結果: # # timeout!

（4）在線程上運行協程：在任何協程中直接調用?blocking_io()?將會在調用期間阻塞事件循環，導致額外的 1 秒運行時間。 而通過改用?asyncio.to_thread()，我們可以在不同的線程中運行它從而不會阻塞事件循環。

asyncio.to_thread(func,?/,?*args,?**kwargs)：在不同的線程中異步地運行函數?func

import asyncio import asyncio, timedef blocking_io():print(f"start blocking_io at {time.strftime('%X')}")time.sleep(1)print(f"blocking_io complete at {time.strftime('%X')}")async def main():print(f"started main at {time.strftime('%X')}")await asyncio.gather(asyncio.to_thread(blocking_io),asyncio.sleep(1))# 如果有更多的任務，可以用list解包的方式放入值# await asyncio.gather(*[tasks])print(f"finished main at {time.strftime('%X')}")asyncio.run(main())"""運行結果：started main at 10:33:33 start blocking_io at 10:33:33blocking_io complete at 10:33:34 finished main at 10:33:34"""

（5）task對象（任務對象）：常用方法，使用高層級的asyncio.create_task()?函數來創建 Task 對象，也可用低層級的?loop.create_task()?或?ensure_future()?函數。不建議手動實例化 Task 對象。

task.cancel(): 取消任務

task.cancelled(): 如果task被取消，返回True

task,done(): 如果task已完成返回True

task.result(): 返回task執行的結果

task.exception(): 返回task異常

task.add_done_callback(callback,?*,?context=None): task執行結束后增加的回調

task.get_coro()：返回由 task?包裝的協程對象

task.get_name()：返回 Task 的名稱

task.set_name(value)：設置 Task 的名稱。

二、曾經遇見的一個面試題

題目描述：用協程的方式，打印出時間，要求體現并發效果

import asyncio import timeasync def job1():await asyncio.sleep(1)print("job1執行結束")async def job2():await asyncio.sleep(3)print("job2執行結束")async def main():print(f"{time.strftime('%X')}")await asyncio.gather(*[job1(), job2()])print(f"{time.strftime('%X')}")""" 這樣寫也行，task對象可以有更多操作async def main():task1 = asyncio.create_task(job1())task2 = asyncio.create_task(job2())print(f"{time.strftime('%X')}")await asyncio.gather(*[task1,task2])print(f"{time.strftime('%X')}")"""asyncio.run(main())"""執行結果：11:11:33 job1執行結束 job2執行結束 11:11:36"""

總結

以上是生活随笔為你收集整理的python异步编程asyncio的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。