Python協(xié)程asyncio模塊的演變及高級(jí)用法

Python協(xié)程及asyncio基礎(chǔ)知識(shí)

協(xié)程(coroutine)也叫微線程，是實(shí)現(xiàn)多任務(wù)的另一種方式，是比線程更小的執(zhí)行單元，一般運(yùn)行在單進(jìn)程和單線程上。因?yàn)樗詭PU的上下文，它可以通過簡(jiǎn)單的事件循環(huán)切換任務(wù)，比進(jìn)程和線程的切換效率更高，這是因?yàn)檫M(jìn)程和線程的切換由操作系統(tǒng)進(jìn)行。

Python實(shí)現(xiàn)協(xié)程的主要借助于兩個(gè)庫(kù)：asyncio和gevent。由于asyncio已經(jīng)成為python的標(biāo)準(zhǔn)庫(kù)了無需pip安裝即可使用，這意味著asyncio作為Python原生的協(xié)程實(shí)現(xiàn)方式會(huì)更加流行。本文僅會(huì)介紹asyncio模塊。如果大家對(duì)gevent也有需求，請(qǐng)留言，我會(huì)單獨(dú)寫篇文章介紹這個(gè)庫(kù)的使用。

asyncio 是從Python3.4引入的標(biāo)準(zhǔn)庫(kù)，直接內(nèi)置了對(duì)協(xié)程異步IO的支持。asyncio 的編程模型本質(zhì)是一個(gè)消息循環(huán)，我們一般先定義一個(gè)協(xié)程函數(shù)(或任務(wù)), 從 asyncio 模塊中獲取事件循環(huán)loop，然后把需要執(zhí)行的協(xié)程任務(wù)(或任務(wù)列表)扔到 loop中執(zhí)行，就實(shí)現(xiàn)了異步IO。

定義協(xié)程函數(shù)及執(zhí)行方法的演變

在最早的Python 3.4中，協(xié)程函數(shù)是通過@asyncio.coroutine 和 yeild from 實(shí)現(xiàn)的, 如下所示。

 import asyncio
 
 @asyncio.coroutine
 def func1(i):
     print("協(xié)程函數(shù){}馬上開始執(zhí)行。".format(i))
     yield from asyncio.sleep(2)
     print("協(xié)程函數(shù){}執(zhí)行完畢!".format(i))
 
 if __name__ == '__main__':
     # 獲取事件循環(huán)
     loop = asyncio.get_event_loop()
 
     # 執(zhí)行協(xié)程任務(wù)
     loop.run_until_complete(func1(1))
 
     # 關(guān)閉事件循環(huán)
     loop.close()

這里我們定義了一個(gè)func1的協(xié)程函數(shù)，我們可以使用asyncio.iscoroutinefunction來驗(yàn)證。定義好協(xié)程函數(shù)后，我們首先獲取事件循環(huán)loop，使用它的run_until_complete方法執(zhí)行協(xié)程任務(wù)，然后關(guān)閉loop。

 print(asyncio.iscoroutinefunction(func1(1))) # True

Python 3.5以后引入了async/await 語(yǔ)法定義協(xié)程函數(shù)，代碼如下所示。每個(gè)協(xié)程函數(shù)都以async聲明，以區(qū)別于普通函數(shù)，對(duì)于耗時(shí)的代碼或函數(shù)我們使用await聲明，表示碰到等待時(shí)掛起，以切換到其它任務(wù)。

 import asyncio
 
 # 這是一個(gè)協(xié)程函數(shù)
 async def func1(i):
     print("協(xié)程函數(shù){}馬上開始執(zhí)行。".format(i))
     await asyncio.sleep(2)
     print("協(xié)程函數(shù){}執(zhí)行完畢!".format(i))
 
 if __name__ == '__main__':
     # 獲取事件循環(huán)
     loop = asyncio.get_event_loop()
 
     # 執(zhí)行協(xié)程任務(wù)
     loop.run_until_complete(func1(1))
 
     # 關(guān)閉事件循環(huán)
     loop.close()

Python 3.7之前執(zhí)行協(xié)程任務(wù)都是分三步進(jìn)行的，代碼有點(diǎn)冗余。Python 3.7提供了一個(gè)更簡(jiǎn)便的asyncio.run方法，上面代碼可以簡(jiǎn)化為：

 import asyncio
 
 async def func1(i):
     print(f"協(xié)程函數(shù){i}馬上開始執(zhí)行。")
     await asyncio.sleep(2)
     print(f"協(xié)程函數(shù){i}執(zhí)行完畢!")
 
 if __name__ == '__main__':
     asyncio.run(func1(1))

注：Python自3.6版本起可以使用f-string來對(duì)字符串進(jìn)行格式化了，相當(dāng)于format函數(shù)的簡(jiǎn)化版。

創(chuàng)建協(xié)程任務(wù)的演變

前面的演示案例中，我們只執(zhí)行了單個(gè)協(xié)程任務(wù)(函數(shù))。實(shí)際應(yīng)用中，我們先由協(xié)程函數(shù)創(chuàng)建協(xié)程任務(wù)，然后把它們加入?yún)f(xié)程任務(wù)列表，最后一起交由事件循環(huán)執(zhí)行。

根據(jù)協(xié)程函數(shù)創(chuàng)建協(xié)程任務(wù)有多種方法，其中最新的是Python 3.7版本提供的asyncio.create_task方法，如下所示：

 # 方法1：使用ensure_future方法。future代表一個(gè)對(duì)象，未執(zhí)行的任務(wù)。
 task1 = asyncio.ensure_future(func1(1))
 task2 = asyncio.ensure_future(func1(2))
 
 # 方法2：使用loop.create_task方法
 task1 = loop.create_task(func1(1))
 task2 = loop.create_task(func1(2))
 
 # 方法3：使用Python 3.7提供的asyncio.create_task方法
 task1 = asyncio.create_task(func1(1))
 task2 = asyncio.create_task(func1(2))

創(chuàng)建多個(gè)協(xié)程任務(wù)列表后，我們還要使用asyncio.wait方法收集協(xié)程任務(wù)，并交由事件循環(huán)處理執(zhí)行。

 import asyncio
 
 async def func1(i):
     print(f"協(xié)程函數(shù){i}馬上開始執(zhí)行。")
     await asyncio.sleep(2)
     print(f"協(xié)程函數(shù){i}執(zhí)行完畢!")
 
 
 async def main():
     tasks = []
     # 創(chuàng)建包含4個(gè)協(xié)程任務(wù)的列表
     for i in range(1, 5):
         tasks.append(asyncio.create_task(func1(i)))
         
     await asyncio.wait(tasks)
 
 if __name__ == '__main__':
     asyncio.run(main())

執(zhí)行效果如下所示，你會(huì)發(fā)現(xiàn)4個(gè)協(xié)程任務(wù)并不是按順序執(zhí)行的。

對(duì)于收集多個(gè)協(xié)程任務(wù)，Python還提供了新的asyncio.gather方法，它的作用asyncio.wait方法類似，但更強(qiáng)大。如果列表中傳入的不是create_task方法創(chuàng)建的協(xié)程任務(wù)，它會(huì)自動(dòng)將函數(shù)封裝成協(xié)程任務(wù)，如下所示：

 import asyncio
 
 async def func1(i):
     print(f"協(xié)程函數(shù){i}馬上開始執(zhí)行。")
     await asyncio.sleep(2)
     print(f"協(xié)程函數(shù){i}執(zhí)行完畢!")
 
 async def main():
     tasks = []
     for i in range(1, 5):
         # 這里未由協(xié)程函數(shù)創(chuàng)建協(xié)程任務(wù)
         tasks.append(func1(i))
         
     # 注意這里*號(hào)。gather自動(dòng)將函數(shù)列表封裝成了協(xié)程任務(wù)。
     await asyncio.gather(*tasks)
 
 if __name__ == '__main__':
     asyncio.run(main())

獲取協(xié)程任務(wù)執(zhí)行結(jié)果

是的，gather方法有將函數(shù)封裝成協(xié)程任務(wù)的能力，但這還并不是兩者最主要的區(qū)別作用。兩者更大的區(qū)別在協(xié)程任務(wù)執(zhí)行完畢后對(duì)于返回結(jié)果的處理上。通常獲取任務(wù)執(zhí)行結(jié)果通常對(duì)于一個(gè)程序至關(guān)重要，因此我們有必要花更多時(shí)間詳細(xì)了解這兩個(gè)方法的使用。

asyncio.wait 會(huì)返回兩個(gè)值：done 和 pending，done 為已完成的協(xié)程任務(wù)列表，pending 為超時(shí)未完成的協(xié)程任務(wù)類別，需通過task.result()方法可以獲取每個(gè)協(xié)程任務(wù)返回的結(jié)果；而asyncio.gather 返回的是所有已完成協(xié)程任務(wù)的 result，不需要再進(jìn)行調(diào)用或其他操作，就可以得到全部結(jié)果。

我們來看兩個(gè)示例。現(xiàn)在修改我們的協(xié)程函數(shù)，通過return給它增加一個(gè)返回值。

通過asyncio.wait獲取協(xié)程任務(wù)執(zhí)行結(jié)果

 import asyncio
 
 async def func1(i):
     print(f"協(xié)程函數(shù){i}馬上開始執(zhí)行。")
     await asyncio.sleep(2)
     return i
 
 async def main():
     tasks = []
     for i in range(1, 5):
         tasks.append(asyncio.create_task(func1(i)))
         
     # 獲取任務(wù)執(zhí)行結(jié)果。
     done, pending = await asyncio.wait(tasks)
     for task in done:
         print(f"執(zhí)行結(jié)果: {task.result()}")
 
 if __name__ == '__main__':
     asyncio.run(main())

執(zhí)行結(jié)果如下所示。你可以看到協(xié)程任務(wù)執(zhí)行結(jié)果并不是按任務(wù)添加的順序返回的。

通過asyncio.gather獲取協(xié)程任務(wù)執(zhí)行結(jié)果

繼續(xù)修改我們的代碼：

 #-*- coding:utf-8 -*-
 import asyncio
 
 async def func1(i):
     print(f"協(xié)程函數(shù){i}馬上開始執(zhí)行。")
     await asyncio.sleep(2)
     return i
 
 async def main():
     tasks = []
     for i in range(1, 5):
         tasks.append(func1(i))
 
     results = await asyncio.gather(*tasks)
     for result in results:
         print(f"執(zhí)行結(jié)果: {result}")
 
 if __name__ == '__main__':
     asyncio.run(main())

執(zhí)行結(jié)果如下所示。協(xié)程任務(wù)執(zhí)行結(jié)果與任務(wù)添加順序完全一致。

現(xiàn)在你知道gather和wait方法的真正區(qū)別了嗎？

• gather具有把普通協(xié)程函數(shù)包裝成協(xié)程任務(wù)的能力，wait沒有。wait只能接收包裝后的協(xié)程任務(wù)列表做參數(shù)。
• 兩者返回值不一樣，wait返回的是已完成和未完成任務(wù)的列表，而gather直接返回協(xié)程任務(wù)執(zhí)行結(jié)果。
• gather返回的任務(wù)執(zhí)行結(jié)果是有序的，wait方法獲取的結(jié)果是無序的。

asyncio高級(jí)使用方法

給任務(wù)添加回調(diào)函數(shù)

我們還可以給每個(gè)協(xié)程任務(wù)通過add_done_callback的方法給單個(gè)協(xié)程任務(wù)添加回調(diào)函數(shù)，如下所示：

 #-*- coding:utf-8 -*-
 import asyncio
 
 async def func1(i):
     print(f"協(xié)程函數(shù){i}馬上開始執(zhí)行。")
     await asyncio.sleep(2)
     return i
 
 # 回調(diào)函數(shù)
 def callback(future):
     print(f"執(zhí)行結(jié)果:{future.result()}")
 
 async def main():
     tasks = []
     for i in range(1, 5):
         task = asyncio.create_task(func1(i))
         
         # 注意這里，增加回調(diào)函數(shù)
         task.add_done_callback(callback)
         tasks.append(task)
 
     await asyncio.wait(tasks)
 
 if __name__ == '__main__':
     asyncio.run(main())

設(shè)置任務(wù)超時(shí)

很多協(xié)程任務(wù)都是很耗時(shí)的，當(dāng)你使用wait方法收集協(xié)程任務(wù)時(shí)，可通過timeout選項(xiàng)設(shè)置任務(wù)切換前單個(gè)任務(wù)最大等待時(shí)間長(zhǎng)度，如下所示：

  # 獲取任務(wù)執(zhí)行結(jié)果，如下所示：
  done,pending = await asyncio.wait(tasks, timeout=10)

自省

• asyncio.current_task: 返回當(dāng)前運(yùn)行的Task實(shí)例，如果沒有正在運(yùn)行的任務(wù)則返回 None。如果 loop 為 None 則會(huì)使用 get_running_loop()獲取當(dāng)前事件循環(huán)。
• asyncio.all_tasks: 返回事件循環(huán)所運(yùn)行的未完成的Task對(duì)象的集合。

以上就是Python協(xié)程asyncio模塊的演變及高級(jí)用法的詳細(xì)內(nèi)容，更多關(guān)于Python協(xié)程asyncio模塊的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！