python asyncio 協(xié)程庫的使用

更新時(shí)間：2021年01月21日 10:51:05 作者：Kevin Wang

這篇文章主要介紹了python asyncio 協(xié)程庫的使用，幫助大家更好的理解和使用python，感興趣的朋友可以了解下

asyncio 是 python 力推多年的攜程庫，與其線程庫相得益彰，更輕量，并且協(xié)程可以訪問同一進(jìn)程中的變量，不需要進(jìn)程間通信來傳遞數(shù)據(jù)，所以使用起來非常順手。

asyncio 官方文檔寫的非常簡練和有效，半小時(shí)內(nèi)可以學(xué)習(xí)和測試完，下面為我的一段 HelloWrold，感覺可以更快速的幫你認(rèn)識協(xié)程。

定義協(xié)程

import asyncio
import time


async def say_after(delay, what):
  await asyncio.sleep(delay)
  print(what)

async 關(guān)鍵字用來聲明一個(gè)協(xié)程函數(shù)，這種函數(shù)不能直接調(diào)用，會拋出異常。正確的調(diào)用姿勢有：

await 協(xié)程()
await asyncio.gather(協(xié)程1()， 協(xié)程2())
await asyncio.waite([協(xié)程1()， 協(xié)程2()])
asyncio.create_task(協(xié)程())

await 阻塞式調(diào)用協(xié)程

先來測試前 3 種 await 的方式：

async def main1():
  # 直接 await，順序執(zhí)行
  await say_after(2, "2s")
  await say_after(1, "1s")


async def main2():
  # 使用 gather，并發(fā)執(zhí)行
  await asyncio.gather(say_after(2, "2s"), say_after(1, "1s"))


async def main3():
  # 使用 wait，簡單等待
  # 3.8 版后已廢棄: 如果 aws 中的某個(gè)可等待對象為協(xié)程，它將自動作為任務(wù)加入日程。直接向 wait() 傳入?yún)f(xié)程對象已棄用，因?yàn)檫@會導(dǎo)致 令人迷惑的行為。
  # 3.10 版后移除
  await asyncio.wait([say_after(2, "2s"), say_after(1, "1s")])

python 規(guī)定：調(diào)用協(xié)程可以用 await，但 await 必須在另一個(gè)協(xié)程中 —— 這不死循環(huán)了？不會的，asyncio 提供了多個(gè)能夠最初調(diào)用協(xié)程的入口：

asyncio.get_event_loop().run_until_complete(協(xié)程)
asyncio.run(協(xié)程)

封裝一個(gè)計(jì)算時(shí)間的函數(shù)，然后把 2 種方式都試一下：

def runtime(entry, func):
  print("-" * 10 + func.__name__)
  start = time.perf_counter()
  entry(func())
  print("=" * 10 + "{:.5f}".format(time.perf_counter() - start))

print("########### 用 loop 入口協(xié)程 ###########")

loop = asyncio.get_event_loop()
runtime(loop.run_until_complete, main1)
runtime(loop.run_until_complete, main2)
runtime(loop.run_until_complete, main3)
loop.close()

print("########### 用 run 入口協(xié)程 ###########")

runtime(asyncio.run, main1)
runtime(asyncio.run, main2)
runtime(asyncio.run, main3)

運(yùn)行結(jié)果：

########### 用 loop 入口協(xié)程 ###########
----------main1
2s
1s
==========3.00923
----------main2
1s
2s
==========2.00600
----------main3
1s
2s
==========2.00612
########### 用 run 入口協(xié)程 ###########
----------main1
2s
1s
==========3.01193
----------main2
1s
2s
==========2.00681
----------main3
1s
2s
==========2.00592

可見，2 種協(xié)程入口調(diào)用方式差別不大

下面，需要明確 2 個(gè)問題：

協(xié)程間的并發(fā)問題：除了 main1 耗時(shí) 3s 外，其他都是 2s，說明 main1 方式串行執(zhí)行 2 個(gè)協(xié)程，其他是并發(fā)執(zhí)行協(xié)程。
協(xié)程是否阻塞父協(xié)程/父進(jìn)程的問題：上述測試都使用了 await，即等待協(xié)程執(zhí)行完畢后再繼續(xù)往下走，所以都是阻塞式的，主進(jìn)程都在此等待協(xié)程的執(zhí)行完。—— 那么如何才能不阻塞父協(xié)程呢？不加 await 行么？ —— 上面 3 種方式都不行！
下面介紹可以不阻塞主協(xié)程的方式。

task 實(shí)現(xiàn)更靈活的協(xié)程

一切都在代碼中：

# 驗(yàn)證 task 啟動協(xié)程是立即執(zhí)行的
async def main4():
  # create_task() Python 3.7 中被加入
  task1 = asyncio.create_task(say_after(2, "2s"))
  task2 = asyncio.create_task(say_after(1, "1s"))
  # 創(chuàng)建任務(wù)后會立即開始執(zhí)行，后續(xù)可以用 await 來等待其完成后再繼續(xù)，也可以被 cancle
  await task1 # 等待 task1 執(zhí)行完，其實(shí)返回時(shí) 2 個(gè)task 都已經(jīng)執(zhí)行完
  print("--") # 最后才會被打印，因?yàn)?2 個(gè)task 都已經(jīng)執(zhí)行完
  await task2
  # 這里是等待所有 task 結(jié)束才繼續(xù)運(yùn)行。


# 驗(yàn)證父協(xié)程與子協(xié)程的關(guān)閉關(guān)系
async def main5():
  task1 = asyncio.create_task(say_after(2, "2s"))
  task2 = asyncio.create_task(say_after(1, "1s"))
  # 如果不等待，函數(shù)會直接 return，main5 協(xié)程結(jié)束，task1/2 子協(xié)程也結(jié)束，所以看不到打印
  # 此處等待 1s，則會只看到 1 個(gè)，等待 >2s，則會看到 2 個(gè) task 的打印
  await asyncio.sleep(2)


# python3.8 后 python 為 asyncio 的 task 增加了很多功能：
# get/set name、獲取正在運(yùn)行的 task、cancel 功能
# 驗(yàn)證 task 的 cancel() 功能
async def cancel_me(t):
  # 定義一個(gè)可處理 CancelledError 的協(xié)程
  print("cancel_me(): before sleep")
  try:
    await asyncio.sleep(t)
  except asyncio.CancelledError:
    print("cancel_me(): cancel sleep")
    raise
  finally:
    print("cancel_me(): after sleep")
  return "I hate be canceled"


async def main6():
  async def test(t1, t2):
    task = asyncio.create_task(cancel_me(t1))
    await asyncio.sleep(t2)
    task.cancel() # 會在 task 內(nèi)引發(fā)一個(gè) CancelledError
    try:
      await task
    except asyncio.CancelledError:
      print("main(): cancel_me is cancelled now")
    try:
      print(task.result())
    except asyncio.CancelledError:
      print("main(): cancel_me is cancelled now")

  # 讓其運(yùn)行2s，但在1s時(shí) cancel 它
  await test(2, 1) # await 和 result 時(shí)都會引發(fā) CancelledError
  await test(1, 2) # await 和 result 時(shí)不會引發(fā)，并且 result 會得到函數(shù)的返回值

runtime(asyncio.run, main4)
runtime(asyncio.run, main5)
runtime(asyncio.run, main6)

運(yùn)行結(jié)果：

----------main4
1s
2s
--
==========2.00557
----------main5
1s
2s
==========3.00160
----------main6
cancel_me(): before sleep
cancel_me(): cancel sleep
cancel_me(): after sleep
main(): cancel_me is cancelled now
main(): cancel_me is cancelled now
cancel_me(): before sleep
cancel_me(): after sleep
I hate be canceled
==========3.00924