python進階之協程你瞭解嗎
協程的定義
協程(Coroutine)，又稱微線程，纖程。(協程是一種用戶態的輕量級線程)
作用：在執行 A 函數的時候，可以隨時中斷，去執行 B 函數，然後中斷B函數，繼續執行 A 函數 （可以自動切換），但這一過程並不是函數調用（沒有調用語句），過程很像多線程，然而協程隻有一個線程在執行
通俗的理解：在一個線程中的某個函數，可以在任何地方保存當前函數的一些臨時變量等信息，然後切換到另外一個函數中執行，註意不是通過調用函數的方式做到的，並且切換的次數以及什麼時候再切換到原來的函數都由開發者自己確定
協程和線程差異
在實現多任務時, 線程切換從系統層面遠不止保存和恢復 CPU上下文這麼簡單。 操作系統為瞭程序運行的高效性每個線程都有自己緩存Cache等等數據，操作系統還會幫你做這些數據的恢復操作。 所以線程的切換非常耗性能。但是協程的切換隻是單純的操作CPU的上下文，所以一秒鐘切換個上百萬次系統都抗的住。
協程的標準
必須在隻有一個單線程裡實現並發
- 修改共享數據不需加鎖
- 用戶程序裡自己保存多個控制流的上下文棧
- 一個協程遇到 IO 操作自動切換到其它協程
協程的優點
- 由於自身帶有上下文和棧，無需線程上下文切換的開銷,屬於程序級別的切換，操作系統完全感知不到，因而更加輕量級
- 無需原子操作的鎖定及同步的開銷
- 方便切換控制流，簡化編程模型
- 單線程內就可以實現並發的效果，最大限度地利用 cpu，且可擴展性高，成本低
協程的缺點
- 無法利用多核資源：協程的本質是個單線程,它不能同時將 單個 CPU 的多個核用上,協程需要和進程配合才能運行在多 CPU 上
- 進行阻塞（Blocking）操作（如 IO 時）會阻塞掉整個程序
- 計算型的操作，利用協程來回切換執行，沒有任何意義，來回切換並保存狀態 反倒會降低性能。
python中實現協程的方式
greenlet，是一個第三方模塊，用於實現協程代碼（Gevent協程就是基於greenlet實現）yield，生成器，借助生成器的特點也可以實現協程代碼。asyncio，在Python3.4中引入的模塊用於編寫協程代碼。async & awiat，在Python3.5中引入的兩個關鍵字，結合asyncio模塊可以更方便的編寫協程代碼（推薦）。
async&await關鍵字
協程實現的方式有多種，目前最流行的方式就是async&await，其他的方式瞭解即可，本文就介紹最流行的一種方式
使用協程需要瞭解2個，事件循環和定義協程函數
事件循環
事件循環是一種處理多並發量的有效方式，可以理解為死循環，循環過程中去檢測並執行某些代碼，我們來看下面的偽代碼
任務列表 = [任務1， 任務2， 任務3............]
while True:
可執行的任務列表，已完成的任務列表 = 去任務列表中檢查所有的任務，將'可執行'和'已完成'的任務返回
for 就緒任務 in 可執行的任務列表：
執行已就緒的任務
for 已完成的任務 in 已完成的任務列表：
在任務列表中移除 已完成的任務
如果任務列表中的任務都已完成，則終止循環
上面的偽代碼的意思：獲取事件循環中，然後不斷監聽任務列表，有任務就執行，執行完成的任務就移除，直到任務列表中的所有任務都完成，終止循環
使用事件循環的好處：使得程序員不用控制任務的添加、刪除和事件的控制
代碼中的寫法如下：
import asyncio # 獲取事件循環 loop = asyncio.get_event_loop() # 將任務放到`任務列表`，監聽事件循環 loop.run_until_complete(任務) # 關閉事件 loop.close()
協程函數和協程對象
想要定義協程函數，格式：async def 函數名協程對象：執行協程函數()得到的協程對象
# 定義協程函數
async def func():
pass
# 創建協程對象
result = func()
註意：執行協程函數，創建協程對象，函數代碼是不會運行的，如果想要運行協程函數的內部代碼，必須要將協程對象交給事件循環來處理，看如下代碼
import asyncio
async def func():
print("你好")
result = func()
# 方式1
loop = asyncio.get_event_loop()
loop.run_until_complete(result)
# 方式2
asyncio.run(result) # python3.7寫法
await
await是一個隻能在協程函數中使用的關鍵字，用於遇到IO操作時掛起 當前協程（任務），當前協程（任務）掛起過程中 事件循環可以去執行其他的協程（任務），當前協程IO處理完成時，可以再次切換回來執行await之後的代碼。
舉例：我們創建瞭2個任務，一個下載圖片，一個下載視頻，我們先執行下載圖片任務，這時遇到io操作，正常情況下會等待圖片下載完，但await可以先掛起下載圖片任務，然後自動切換到下載視頻任務
使用方法：await + 可等待對象（協程對象、Future對象、Task對象）
案例1
import asyncio
async def func():
print("執行協程函數內部代碼")
# 遇到IO操作掛起當前協程（任務），等IO操作完成之後再繼續往下執行。
# 當前協程掛起時，事件循環可以去執行其他協程（任務）。
response = await asyncio.sleep(2)
print("IO請求結束，結果為：", response)
result = func()
asyncio.run(result)
案列2
import asyncio
async def others():
print("start") # ④打印start
await asyncio.sleep(2) # ⑤等待耗時2秒，這過程中可以切換到其他協程中去
print("end") # ⑥打印end
return '返回值'
async def func():
print("執行協程函數內部代碼") # ②執行協程函數，打印print代碼
response = await others() # ③等待協程函數others
print(f"io請求結束，結果為{response}") # ⑦等待others結束後打印print語句
if __name__ == '__main__':
asyncio.run(func()) # ①協程函數放入事件循環中運行
上述的所有示例都隻是創建瞭一個任務，即：事件循環的任務列表中隻有一個任務，所以在IO等待時無法演示切換到其他任務效果。在程序想要創建多個任務對象，需要使用Task對象來實現。
Task對象
Tasks用於並發調度協程，通過asyncio.create_task(協程對象)的方式創建Task對象，這樣可以讓協程加入事件循環中等待被調度執行。除瞭使用 asyncio.create_task() 函數以外，還可以用低層級的loop.create_task()或 ensure_future() 函數。不建議手動實例化 Task 對象。
本質上是將協程對象封裝成task對象，並將協程立即加入事件循環，同時追蹤協程的狀態。
註意：asyncio.create_task() 函數在 Python 3.7 中被加入。在 Python 3.7 之前，可以改用低層級的 asyncio.ensure_future() 函數。
案例1
import asyncio
async def func():
print(1)
await asyncio.sleep(2)
print(2)
return "返回值"
async def main():
print("main開始")
# 創建協程，將協程封裝到一個Task對象中並立即添加到事件循環的任務列表中，等待事件循環去執行（默認是就緒狀態）。
task1 = asyncio.create_task(func())
# 創建協程，將協程封裝到一個Task對象中並立即添加到事件循環的任務列表中，等待事件循環去執行（默認是就緒狀態）。
task2 = asyncio.create_task(func())
print("main結束")
# 當執行某協程遇到IO操作時，會自動化切換執行其他任務。
# 此處的await是等待相對應的協程全都執行完畢並獲取結果
ret1 = await task1
ret2 = await task2
print(ret1, ret2)
asyncio.run(main())
案例await+任務列表（用的最多）
import asyncio
async def func():
print(1)
await asyncio.sleep(2)
print(2)
return "返回值"
async def main():
print("main開始")
# 創建協程，將協程封裝到Task對象中並添加到事件循環的任務列表中，等待事件循環去執行（默認是就緒狀態）。
# 在調用
task_list = [asyncio.create_task(func()), asyncio.create_task(func())]
print("main結束")
# 當執行某協程遇到IO操作時，會自動化切換執行其他任務。
# 此處的await是等待所有協程執行完畢，並將所有協程的返回值保存到done
# 如果設置瞭timeout值，則意味著此處最多等待的秒，完成的協程返回值寫入到done中，未完成則寫到pending中。
done, pending = await asyncio.wait(task_list)
print(done)
asyncio.run(main())
註意： asyncio.wait 源碼內部會對列表中的每個協程執行ensure_future從而封裝為Task對象，所以在和wait配合使用時task_list的值為[func(),func()] 也是可以的。
asyncio.Future對象
asyncio中的Future對象是一個相對更偏向底層的可對象，通常我們不會直接用到這個對象，而是直接使用Task對象來完成任務的並和狀態的追蹤。（ Task 是 Futrue的子類 ）
Future為我們提供瞭異步編程中的 最終結果 的處理（Task類也具備狀態處理的功能）
案例1
async def main():
# 獲取當前事件循環
loop = asyncio.get_running_loop()
# 創建一個任務（Future對象），這個任務什麼都不幹。
fut = loop.create_future()
# 等待任務最終結果（Future對象），沒有結果則會一直等下去。
await fut
asyncio.run(main())
結果就是程序一直在等待，無法結束
案例2
import asyncio
async def set_after(fut):
await asyncio.sleep(2)
fut.set_result("666")
async def main():
# 獲取當前事件循環
loop = asyncio.get_running_loop()
# 創建一個任務（Future對象），沒綁定任何行為，則這個任務永遠不知道什麼時候結束。
fut = loop.create_future()
# 創建一個任務（Task對象），綁定瞭set_after函數，函數內部在2s之後，會給fut賦值。
# 即手動設置future任務的最終結果，那麼fut就可以結束瞭。
await loop.create_task(set_after(fut))
# 等待 Future對象獲取 最終結果，否則一直等下去
data = await fut
print(data)
asyncio.run(main())
Future對象本身函數進行綁定，所以想要讓事件循環獲取Future的結果，則需要手動設置。而Task對象繼承瞭Future對象，其實就對Future進行擴展，他可以實現在對應綁定的函數執行完成之後，自動執行set_result，從而實現自動結束。
雖然，平時使用的是Task對象，但對於結果的處理本質是基於Future對象來實現的。
futures.Future對象
在Python的concurrent.futures模塊中也有一個Future對象，這個對象是基於線程池和進程池實現異步操作時使用的對象。
import time
from concurrent.futures import Future
from concurrent.futures.thread import ThreadPoolExecutor
from concurrent.futures.process import ProcessPoolExecutor
def func(value):
time.sleep(1)
print(value)
pool = ThreadPoolExecutor(max_workers=5)
# 或 pool = ProcessPoolExecutor(max_workers=5)
for i in range(10):
fut = pool.submit(func, i)
print(fut)
兩個Future對象是不同的，他們是為不同的應用場景而設計，例如：concurrent.futures.Future不支持await語法 等。
在Python提供瞭一個將futures.Future 對象包裝成asyncio.Future對象的函數 asynic.wrap_future。
接下裡你肯定問：為什麼python會提供這種功能？
其實，一般在程序開發中我們要麼統一使用 asycio 的協程實現異步操作、要麼都使用進程池和線程池實現異步操作。但如果 協程的異步和 進程池/線程池的異步 混搭時，那麼就會用到此功能瞭。
import time
import asyncio
import concurrent.futures
def func1():
# 某個耗時操作
time.sleep(2)
return "OK"
async def main():
loop = asyncio.get_running_loop()
# 方式1. Run in the default loop's executor ( 默認ThreadPoolExecutor )
# 第一步：內部會先調用 ThreadPoolExecutor 的 submit 方法去線程池中申請一個線程去執行func1函數，並返回一個concurrent.futures.Future對象
# 第二步：調用asyncio.wrap_future將concurrent.futures.Future對象包裝為asycio.Future對象。
# 因為concurrent.futures.Future對象不支持await語法，所以需要包裝為 asycio.Future對象 才能使用。
fut = loop.run_in_executor(None, func1)
result = await fut
print('default thread pool', result)
# 方式2. Run in a custom thread pool:
# with concurrent.futures.ThreadPoolExecutor() as pool:
# result = await loop.run_in_executor(
# pool, func1)
# print('custom thread pool', result)
# 方式3. Run in a custom process pool:
# with concurrent.futures.ProcessPoolExecutor() as pool:
# result = await loop.run_in_executor(
# pool, func1)
# print('custom process pool', result)
asyncio.run(main())
應用場景：當項目以協程式的異步編程開發時，如果要使用一個第三方模塊，而第三方模塊不支持協程方式異步編程時，就需要用到這個功能，例如requests模塊：
import asyncio
import requests
async def download_image(url):
# 發送網絡請求，下載圖片（遇到網絡下載圖片的IO請求，自動化切換到其他任務）
print("開始下載:", url)
loop = asyncio.get_event_loop()
# requests模塊默認不支持異步操作，所以就使用線程池來配合實現瞭。
future = loop.run_in_executor(None, requests.get, url)
response = await future
print('下載完成')
# 圖片保存到本地文件
file_name = url.rsplit('_')[-1]
with open(file_name, mode='wb') as file_object:
file_object.write(response.content)
if __name__ == '__main__':
url_list = [
'https://www3.autoimg.cn/newsdfs/g26/M02/35/A9/120x90_0_autohomecar__ChsEe12AXQ6AOOH_AAFocMs8nzU621.jpg',
'https://www2.autoimg.cn/newsdfs/g30/M01/3C/E2/120x90_0_autohomecar__ChcCSV2BBICAUntfAADjJFd6800429.jpg',
'https://www3.autoimg.cn/newsdfs/g26/M0B/3C/65/120x90_0_autohomecar__ChcCP12BFCmAIO83AAGq7vK0sGY193.jpg'
]
tasks = [download_image(url) for url in url_list]
loop = asyncio.get_event_loop()
loop.run_until_complete( asyncio.wait(tasks) )
異步迭代器
什麼是異步迭代器？
實現瞭__aiter__()和__anext__()方法的對象。__anext__ 必須返回一個 awaitable 對象。async for 會處理異步迭代器的__anext__()方法所返回的可等待對象，直到其引發一個 StopAsyncIteration 異常。
什麼是異步可迭代對象？
可在 async for 語句中被使用的對象。必須通過它的 __aiter__() 方法返回一個 asynchronous iterator。
import asyncio
class Reader(object):
""" 自定義異步迭代器（同時也是異步可迭代對象） """
def __init__(self):
self.count = 0
async def readline(self):
# await asyncio.sleep(1)
self.count += 1
if self.count == 100:
return None
return self.count
def __aiter__(self):
return self
async def __anext__(self):
val = await self.readline()
if val == None:
raise StopAsyncIteration
return val
async def func():
# 創建異步可迭代對象
async_iter = Reader()
# async for 必須要放在async def函數內，否則語法錯誤。
async for item in async_iter:
print(item)
asyncio.run(func())
異步迭代器其實沒什麼太大的作用，隻是支持瞭async for語法而已。
異步上下文管理器
此種對象通過定義 __aenter__() 和 __aexit__() 方法來對 async with 語句中的環境進行控制
import asyncio
class AsyncContextManager:
def __init__(self):
self.conn = None
async def do_something(self):
# 異步操作數據庫
return 666
async def __aenter__(self):
# 異步鏈接數據庫
self.conn = await asyncio.sleep(1)
return self
async def __aexit__(self, exc_type, exc, tb):
# 異步關閉數據庫鏈接
await asyncio.sleep(1)
async def func():
async with AsyncContextManager() as f:
result = await f.do_something()
print(result)
asyncio.run(func())
這個異步的上下文管理器還是比較有用的，平時在開發過程中 打開、處理、關閉 操作時，就可以用這種方式來處理。
uvloop
uvloop是 asyncio 中的事件循環的替代方案，替換後可以使得asyncio性能提高。事實上，uvloop要比nodejs、gevent等其他python異步框架至少要快2倍，性能可以比肩Go語言。
安裝uvloop
pip3 install uvloop
在項目中想要使用uvloop替換asyncio的事件循環也非常簡單，隻要在代碼中這麼做就行。
import asyncio import uvloop asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()) # 編寫asyncio的代碼，與之前寫的代碼一致。 # 內部的事件循環自動化會變為uvloop asyncio.run(...)
註意：知名的asgi uvicorn內部就是使用的uvloop的事件循環。
異步redis
當通過python去操作redis時，鏈接、設置值、獲取值 這些都涉及網絡IO請求，使用asycio異步的方式可以在IO等待時去做一些其他任務，從而提升性能。
安裝Python異步操作redis模塊
pip3 install aioredis
案例:連接多個redis做操作（遇到IO會切換其他任務，提供瞭性能）
import asyncio
import aioredis
async def execute(address, password):
print("開始執行", address)
# 網絡IO操作：先去連接 77.95.4.197:6379，遇到IO則自動切換任務，去連接77.95.4.198:6379
redis = await aioredis.create_redis_pool(address, password=password)
# 網絡IO操作：遇到IO會自動切換任務
await redis.hmset_dict('car', key1=1, key2=2, key3=3)
# 網絡IO操作：遇到IO會自動切換任務
result = await redis.hgetall('car', encoding='utf-8')
print(result)
redis.close()
# 網絡IO操作：遇到IO會自動切換任務
await redis.wait_closed()
print("結束", address)
task_list = [
execute('redis://77.95.4.197:6379', "123456"),
execute('redis://77.95.4.198:6379', "123456")
]
asyncio.run(asyncio.wait(task_list))
異步MySQL
當通過python去操作MySQL時，連接、執行SQL、關閉都涉及網絡IO請求，使用asycio異步的方式可以在IO等待時去做一些其他任務，從而提升性能。
安裝Python異步操作redis模塊
pip3 install aiomysql
案例
import asyncio
import aiomysql
async def execute(host, password):
print("開始", host)
# 網絡IO操作：先去連接 77.95.40.197，遇到IO則自動切換任務，去連接77.95.40.198:6379
conn = await aiomysql.connect(host=host, port=3306, user='root', password=password, db='mysql')
# 網絡IO操作：遇到IO會自動切換任務
cur = await conn.cursor()
# 網絡IO操作：遇到IO會自動切換任務
await cur.execute("SELECT Host,User FROM user")
# 網絡IO操作：遇到IO會自動切換任務
result = await cur.fetchall()
print(result)
# 網絡IO操作：遇到IO會自動切換任務
await cur.close()
conn.close()
print("結束", host)
task_list = [
execute('77.95.40.197', "123456"),
execute('77.95.40.198', "123456")
]
asyncio.run(asyncio.wait(task_list))
爬蟲
在編寫爬蟲應用時，需要通過網絡IO去請求目標數據，這種情況適合使用異步編程來提升性能，接下來我們使用支持異步編程的aiohttp模塊來實現。
安裝aiohttp模塊
pip3 install aiohttp
案例
import aiohttp
import asyncio
async def fetch(session, url):
print(f"發送請求：{url}")
async with session.get(url, verify_ssl=False) as response:
text = await response.text()
print("得到結果：", url, len(text))
async def main():
async with aiohttp.ClientSession() as session:
url_list = ["http://www.baidu.com", "http://www.taobao.com", "http://www.jd.com"]
tasks = [asyncio.create_task(fetch(session, url)) for url in url_list]
await asyncio.wait(tasks)
if __name__ == '__main__':
asyncio.run(main())
總結
本篇文章就到這裡瞭，希望能夠給你帶來幫助，也希望您能夠多多關註WalkonNet的更多內容!
推薦閱讀：
- python中Task封裝協程的知識點總結
- Python協程asyncio模塊的演變及高級用法
- Python協程asyncio異步編程筆記分享
- python協程與 asyncio 庫詳情
- python asyncio 協程庫的使用