Python中的并發處理之asyncio包使用的詳解

2020-01-04 15:31:34

字體：大中小

來源：轉載

供稿：網友

導語：本文章記錄了本人在學習Python基礎之控制流程篇的重點知識及個人心得，打算入門Python的朋友們可以來一起學習并交流。

本文重點：

1、了解asyncio包的功能和使用方法；
2、了解如何避免阻塞型調用；
3、學會使用協程避免回調地獄。

一、使用asyncio包做并發編程

1、并發與并行

并發：一次處理多件事。
并行：一次做多件事。
并發用于制定方案，用來解決可能（但未必）并行的問題。并發更好。

2、asyncio概述

了解asyncio的4個特點：

asyncio包使用事件循環驅動的協程實現并發。
適合asyncio API的協程在定義體中必須使用yield from，而不能使用yield。
使用asyncio處理的協程，需在定義體上使用@asyncio.coroutine裝飾。裝飾的功能在于凸顯協程，同時當協程不產出值，協程會被垃圾回收。
Python3.4起，asyncio包只直接支持TCP和UDP協議。如果想使用asyncio實現HTTP客戶端和服務器時，常使用aiohttp包。

在協程中使用yield from需要注意兩點：

使用yield froml鏈接的多個協程最終必須由不是協程的調用方驅動，調用方顯式或隱式在最外層委派生成器上調用next()函數或 .send()方法。
鏈條中最內層的子生成器必須是簡單的生成器（只使用yield）或可迭代的對象。

但在asyncio包的API中使用yield from還需注意兩個細節：

asyncio包中編寫的協程鏈條始終通過把最外層委派生成器傳給asyncio包API中的某個函數驅動，例如loop.run_until_complete()。即不通過調用next()函數或 .send()方法驅動協程。
編寫的協程鏈條最終通過yield from把職責委托給asyncio包中的某個協程函數或協程方法。即最內層的子生成器是庫中真正執行I/O操作的函數，而不是我們自己編寫的函數。

實例——通過asyncio包和協程以動畫形式顯示文本式旋轉指針：

import asyncioimport itertoolsimport sys@asyncio.coroutine # 交給 asyncio 處理的協程要使用 @asyncio.coroutine 裝飾def spin(msg):  for char in itertools.cycle('|/-//'):    status = char + ' ' + msg    print(status)    try:      yield from asyncio.sleep(.1) # 使用 yield from asyncio.sleep(.1) 代替 time.sleep(.1)，這樣的休眠不會阻塞事件循環。    except asyncio.CancelledError: # 如果 spin 函數蘇醒后拋出 asyncio.CancelledError 異常，其原因是發出了取消請求，因此退出循環。      break@asyncio.coroutinedef slow_function(): # slow_function 函數是協程，在用休眠假裝進行 I/O 操作時，使用 yield from 繼續執行事件循環。  # 假裝等待I/O一段時間  yield from asyncio.sleep(3) # yield from asyncio.sleep(3) 表達式把控制權交給主循環，在休眠結束后恢復這個協程。  return 42@asyncio.coroutinedef supervisor(): # supervisor 函數也是協程  spinner = asyncio.async(spin('thinking!')) # asyncio.async(...) 函數排定 spin 協程的運行時間，使用一個 Task 對象包裝spin 協程，并立即返回。  print('spinner object:', spinner)  result = yield from slow_function() # 驅動 slow_function() 函數。結束后，獲取返回值。# 同時，事件循環繼續運行，因為slow_function 函數最后使用 yield from asyncio.sleep(3) 表達式把控制權交回給了主循環。  spinner.cancel() # Task 對象可以取消；取消后會在協程當前暫停的 yield 處拋出 asyncio.CancelledError 異常。協程可以捕獲這個異常，也可以延遲取消，甚至拒絕取消。  return resultif __name__ == '__main__':  loop = asyncio.get_event_loop() # 獲取事件循環的引用  result = loop.run_until_complete(supervisor()) # 驅動 supervisor 協程，讓它運行完畢；這個協程的返回值是這次調用的返回值。  loop.close()  print('Answer:', result)

3、線程與協程對比

線程：調度程序在任何時候都能中斷線程。必須記住保留鎖。去保護程序中的重要部分，防止多步操作在執行的過程中中斷，防止數據處于無效狀態。

協程：默認會做好全方位保護，以防止中斷。對協程來說無需保留鎖，在多個線程之間同步操作，協程自身就會同步，因為在任意時刻只有一個協程運行。

4、從期物、任務和協程中產出

在asyncio包中，期物和協程關系緊密，因為可以使用yield from從asyncio.Future對象中產出結果。這意味著，如果foo是協程函數，抑或是返回Future或Task實例的普通函數，那么可以這樣寫：res=yield from foo()。這是asyncio包中很多地方可以互換協程與期物的原因之一。

二、避免阻塞型調用

1、有兩種方法能避免阻塞型調用中止整個應用程序的進程：

在單獨的線程中運行各個阻塞型操作。
把每個阻塞型操作轉換成非阻塞的異步調用。

使用多線程處理大量連接時將耗費過多的內存，故此通常使用回調來實現異步調用。

2、使用Executor對象防止阻塞事件循環：

使用loop.run_in_executor把阻塞的作業（例如保存文件）委托給線程池做。

@asyncio.coroutinedef download_one(cc, base_url, semaphore, verbose):  try:    with (yield from semaphore):      image = yield from get_flag(base_url, cc)  except web.HTTPNotFound:    status = HTTPStatus.not_found    msg = 'not found'  except Exception as exc:    raise FetchError(cc) from exc  else:    loop = asyncio.get_event_loop() # 獲取事件循環對象的引用    loop.run_in_executor(None, # None 使用默認的 TrreadPoolExecutor 實例        save_flag, image, cc.lower() + '.gif') # 傳入可調用對象    status = HTTPStatus.ok    msg = 'OK'  if verbose and msg:    print(cc, msg)  return Result(status, cc)

asyncio 的事件循環背后維護一個 ThreadPoolExecutor 對象，我們可以調用 run_in_executor 方法，把可調用的對象發給它執行。

三、從回調到期物和協程

回調地獄：如果一個操作需要依賴之前操作的結果，那就得嵌套回調。

Python 中的回調地獄：

def stage1(response1):  request2 = step1(response1)  api_call2(request2, stage2)def stage2(response2):  request3 = step2(response2)  api_call3(request3, stage3)def stage3(response3):  step3(response3)api_call1(request1, step1)

使用協程和 yield from 結構做異步編程，無需用回調：

@asyncio.coroutinedef three_stages(request1):  response1 = yield from api_call1()  request2 = step1(response1)  response2 = yield from api_call2(request2)  request3 = step2(response2)  response3 = yield from api_call3(request3)  step3(response3)loop.create_task(three_stages(request1))# 協程不能直接調用，必須用事件循環顯示指定協程的執行時間，或者在其他排定了執行時間的協程中使用 yield from 表達式把它激活

四、使用asyncio包編寫服務器