python協(xié)程庫asyncio(異步io)問題
介紹
- 異步IO:就是發(fā)起一個IO操作(如:網絡請求,文件讀寫等),這些操作一般是比較耗時的,不用等待它結束,可以繼續(xù)做其他事情,結束時會發(fā)來通知。
- 協(xié)程:又稱為微線程,在一個線程中執(zhí)行,執(zhí)行函數時可以隨時中斷,由程序(用戶)自身控制,執(zhí)行效率極高,與多線程比較,沒有切換線程的開銷和多線程鎖機制。
asyncio中幾個重要概念
1.事件循環(huán)
事件循環(huán)是每個 asyncio 應用的核心,管理所有的事件,在整個程序運行過程中不斷循環(huán)執(zhí)行并追蹤事件發(fā)生的順序將它們放在隊列中,空閑時調用相應的事件處理者來處理這些事件。
創(chuàng)建事件循環(huán)
loop = asyncio.get_event_loop()
獲取當前事件循環(huán)。
如果當前 OS 線程沒有設置當前事件循環(huán)并且 set_event_loop() 還沒有被調用,asyncio 將創(chuàng)建一個新的事件循環(huán)并將其設置為當前循環(huán)。
另起一個線程創(chuàng)建事件循環(huán)
from threading import Thread
import asyncio
def start_thread_loop(loop):
asyncio.set_event_loop(loop)
loop.run_forever()
new_loop = asyncio.new_event_loop()
loop_thread = Thread(target=start_thread_loop, args=(new_loop,))
loop_thread.setDaemon(True) # 守護線程
loop_thread.start()
2.Future
Future對象表示尚未完成的計算,還未完成的結果,它和task上沒有本質上的區(qū)別
3.Task
是Future的子類,作用是在運行某個任務的同時可以并發(fā)的運行多個任務。asyncio.Task用于實現協(xié)作式多任務的庫,且Task對象不能用戶手動實例化,通過下面2個函數創(chuàng)建:loop.create_task() 或 asyncio.ensure_future()
- loop.create_task() ,要在定義loop對象之后,調用將方法對象轉化成了task的對象
- asyncio.ensure_future() 直接調用asyncio 的ensure_future() 方法,返回的也是task 對象(我們還沒有聲明 loop 也可以提前定義好 task 對象)
4.async/await 關鍵字
asyncio實現了TCP、UDP、SSL等協(xié)議,async定義一個協(xié)程,await用于掛起阻塞的異步調用接口。對于異步io你需要知道的重點,要注意的是,await語法只能出現在通過async修飾的函數中,否則會報SyntaxError錯誤。而且await后面的對象需要是一個Awaitable,或者實現了相關的協(xié)議。
注意
- 所有需要異步執(zhí)行的函數,都需要asyncio中的輪詢器去輪詢執(zhí)行,如果函數阻塞,輪詢器就會去執(zhí)行下一個函數。所以所有需要異步執(zhí)行的函數都需要加入到這個輪詢器中。
- 若在協(xié)程中需要有延時操作,應該使用 await asyncio.sleep(),而不是使用time.sleep(),因為使用time.sleep()后會釋放GIL,阻塞整個主線程,從而阻塞整個事件循環(huán)。
創(chuàng)建一個協(xié)程
使用async可以定義協(xié)程對象,使用await可以針對耗時的操作進行掛起,就像生成器里的yield一樣,函數讓出控制權。協(xié)程遇到await,事件循環(huán)將會掛起該協(xié)程,執(zhí)行別的協(xié)程,直到其他的協(xié)程也掛起或者執(zhí)行完畢,再進行下一個協(xié)程的執(zhí)行
耗時的操作一般是一些IO操作,例如網絡請求,文件讀取等。我們使用asyncio.sleep函數來模擬IO操作。協(xié)程的目的也是讓這些IO操作異步化。
簡單例子
import asyncio
async def execute(x):
print('Number:', x)
coroutine = execute(1)
print('Coroutine:', coroutine)
print('After calling execute')
loop = asyncio.get_event_loop()
loop.run_until_complete(coroutine)
print('After calling loop')
首先我們引入了 asyncio 這個包,這樣我們才可以使用 async 和 await,然后我們使用 async 定義了一個 execute() 方法,方法接收一個數字參數,方法執(zhí)行之后會打印這個數字。
隨后我們直接調用了這個方法,然而這個方法并沒有執(zhí)行,而是返回了一個 coroutine 協(xié)程對象。隨后我們使用 get_event_loop() 方法創(chuàng)建了一個事件循環(huán) loop,并調用了 loop 對象的 run_until_complete() 方法將協(xié)程注冊到事件循環(huán) loop 中,然后啟動。最后我們才看到了 execute() 方法打印了輸出結果。
可見,async 定義的方法就會變成一個無法直接執(zhí)行的 coroutine 對象,必須將其注冊到事件循環(huán)中才可以執(zhí)行。
進階例子
多個任務,定義一個task列表,使用asyncio.gather(*tasks) 或 asyncio.wait(tasks) 接收
import asyncio
import time
now = lambda: time.time()
"""
asyncio.gather主要集中在收集結果上。它等待一堆task并按給定的順序返回結果。
asyncio.wait等待task。而不是直接給你結果,它提供完成和待處理的任務。你必須手工收集結果。
asyncio.wait(tasks) ps:asyncio.wait([1,2,3]) 也可以使用 asyncio.gather(*tasks) ps: asyncio.gather(1,2,3),前者接受一個task列表,后者接收一堆task。
"""
# 定義一個異步任務
async def do_some_work(x):
print("waiting:", x)
# 模擬io阻塞
await asyncio.sleep(x)
return "Done after {}s".format(x)
async def main(loop):
"""
:param loop: loop.create_task(需要傳進loop參數)
:return: None
"""
coroutine1 = do_some_work(1)
coroutine2 = do_some_work(2)
coroutine3 = do_some_work(4)
# asyncio.ensure_future
tasks = [
asyncio.ensure_future(coroutine1),
asyncio.ensure_future(coroutine2),
asyncio.ensure_future(coroutine3)
]
# loop.create_task(需要傳進loop參數)
# tasks = [
# loop.create_task(coroutine1),
# loop.create_task(coroutine2),
# loop.create_task(coroutine3)
# ]
# 返回 完成的 task object
dones, pendings = await asyncio.wait(tasks)
print(dones, pendings)
for task in dones:
print("Task ret:", task.result())
# 返回 task 方法的 返回值
# results = await asyncio.gather(*tasks)
# for result in results:
# print("Task ret:",result)
start = now()
loop = asyncio.get_event_loop()
loop.run_until_complete(main(loop))
print("Time:", now() - start)
gather和wait 的區(qū)別
把多個協(xié)程注冊進一個事件循環(huán)中的兩種方法
使用方式區(qū)別
1.使用asyncio.wait()
loop = asyncio.get_event_loop() loop.run_until_complete(asyncio.wait(tasks))
2.使用asyncio.gather()
loop = asyncio.get_event_loop() loop.run_until_complete(asyncio.gather(*tasks)) # *接收args參數
接收參數區(qū)別
asyncio.wait
參數必須是list對象 ,list 對象存放多個 task object
用asyncio.ensure_future轉為task對象
tasks=[
asyncio.ensure_future(coroutine1),
asyncio.ensure_future(coroutine2),
asyncio.ensure_future(coroutine3)
]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))
不轉為task對象
loop = asyncio.get_event_loop()
tasks=[
coroutine1,
coroutine2,
coroutine3
]
loop.run_until_complete(asyncio.wait(tasks))
asyncio.gather
必須用 * 來接收 list 對象
tasks=[
asyncio.ensure_future(coroutine1),
asyncio.ensure_future(coroutine2),
asyncio.ensure_future(coroutine3)
]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.gather(*tasks))
返回結果區(qū)別
asyncio.wait
asyncio.wait返回dones和pendings
- dones:表示已經完成的任務
- pendings:表示未完成的任務
我們需要手動去獲取結果
dones, pendings = await asyncio.wait(tasks)
print(dones, pendings)
for task in dones:
print("Task ret:", task.result())
asyncio.gather
它的返回值就是 return的結果 ,不用再task.result() 來獲取
# 返回 task 方法的 返回值
results = await asyncio.gather(*tasks)
for result in results:
print("Task ret:",result)
另 asyncio.wait 帶有控制功能
【控制運行任務數】:運行第一個任務就返回
- FIRST_COMPLETED :第一個任務完全返回
- FIRST_EXCEPTION:產生第一個異常返回
- ALL_COMPLETED:所有任務完成返回 (默認選項)
import asyncio
import random
async def coro(tag):
print(">", tag)
await asyncio.sleep(random.uniform(0.5, 5))
print("<", tag)
return tag
loop = asyncio.get_event_loop()
tasks = [coro(i) for i in range(1, 11)]
# 第一次wait 完成情況
print("Get first result:")
finished, unfinished = loop.run_until_complete(
asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)) # 第一個任務完全返回
for task in finished:
print(task.result())
print("unfinished:", len(unfinished))
# 繼續(xù)第一次未完成任務
print("Get more results in 2 seconds:")
finished2, unfinished2 = loop.run_until_complete(
asyncio.wait(unfinished, timeout=2)) # 超時2s 返回
for task in finished2:
print(task.result())
print("unfinished2:", len(unfinished2))
# 繼續(xù)第2次未完成任務
print("Get all other results:")
finished3, unfinished3 = loop.run_until_complete(asyncio.wait(unfinished2)) # ALL_COMPLETED:所有任務完成返回 (默認項)
for task in finished3:
print(task.result())
loop.close()
動態(tài)添加協(xié)程
很多時候,我們的事件循環(huán)用于注冊協(xié)程,而有的協(xié)程需要動態(tài)的添加到事件循環(huán)中。一個簡單的方式就是使用多線程。當前線程創(chuàng)建一個事件循環(huán),然后在新建一個線程,在新線程中啟動事件循環(huán)。當前線程不會被block
相關函數介紹:
- loop.call_soon_threadsafe() :與 call_soon()類似,等待此函數返回后馬上調用回調函數,返回值是一個 asyncio.Handle 對象,此對象內只有一個方法為 cancel()方法,用來取消回調函數。
- loop.call_soon() : 與call_soon_threadsafe()類似,call_soon_threadsafe() 是線程安全的
- loop.call_later():延遲多少秒后執(zhí)行回調函數
- loop.call_at():在指定時間執(zhí)行回調函數,這里的時間統(tǒng)一使用 loop.time() 來替代 time.sleep()
- asyncio.run_coroutine_threadsafe(): 動態(tài)的加入協(xié)程,參數為一個回調函數和一個loop對象,返回值為future對象,通過future.result()獲取回調函數返回值
動態(tài)添加協(xié)程同步方式通過調用 call_soon_threadsafe()函數,傳入一個回調函數callback和一個位置參數
注意:同步方式,回調函數 more_work()為普通函數
import asyncio
from threading import Thread
import time
now = lambda: time.time()
def start_loop(loop):
asyncio.set_event_loop(loop)
loop.run_forever()
def more_work(x):
print('More work {}'.format(x))
time.sleep(x)
print('Finished more work {}'.format(x))
start = now()
new_loop = asyncio.new_event_loop()
t = Thread(target=start_loop, args=(new_loop,))
t.start()
print('TIME: {}'.format(time.time() - start))
new_loop.call_soon_threadsafe(more_work, 6)
new_loop.call_soon_threadsafe(more_work, 3)
print('here')
啟動上述代碼之后,當前線程不會被block,新線程中會按照順序執(zhí)行call_soon_threadsafe方法注冊的more_work方法, 后者因為time.sleep操作是同步阻塞的,因此運行完畢more_work需要大致6 + 3
異步方式
import asyncio
import time
from threading import Thread
now = lambda: time.time()
def start_loop(loop):
asyncio.set_event_loop(loop)
loop.run_forever()
async def do_some_work(x):
print('Waiting {}'.format(x))
await asyncio.sleep(x)
print('Done after {}s'.format(x))
start = now()
new_loop = asyncio.new_event_loop()
t = Thread(target=start_loop, args=(new_loop,))
t.start()
print('TIME: {}'.format(time.time() - start))
asyncio.run_coroutine_threadsafe(do_some_work(6), new_loop)
asyncio.run_coroutine_threadsafe(do_some_work(4), new_loop)
上述的例子,主線程中創(chuàng)建一個new_loop,然后在另外的子線程中開啟一個無限事件循環(huán)。
主線程通過run_coroutine_threadsafe新注冊協(xié)程對象。這樣就能在子線程中進行事件循環(huán)的并發(fā)操作,同時主線程又不會被block。一共執(zhí)行的時間大概在6s左右。
協(xié)程的停止
future對象有幾個狀態(tài):
PendingRunningDoneCacelled
創(chuàng)建future的時候,task為pending,事件循環(huán)調用執(zhí)行的時候當然就是running,調用完畢自然就是done,如果需要停止事件循環(huán),就需要先把task取消??梢允褂胊syncio.Task獲取事件循環(huán)的task
import asyncio
import time
now = lambda: time.time()
async def do_some_work(x):
print("Waiting:", x)
await asyncio.sleep(x)
return "Done after {}s".format(x)
coroutine1 = do_some_work(1)
coroutine2 = do_some_work(2)
coroutine3 = do_some_work(2)
tasks = [
asyncio.ensure_future(coroutine1),
asyncio.ensure_future(coroutine2),
asyncio.ensure_future(coroutine3),
]
start = now()
loop = asyncio.get_event_loop()
try:
loop.run_until_complete(asyncio.wait(tasks))
except KeyboardInterrupt as e:
print(asyncio.Task.all_tasks())
for task in asyncio.Task.all_tasks():
print(task.cancel())
loop.stop()
loop.run_forever()
finally:
loop.close()
print("Time:", now() - start)
啟動事件循環(huán)之后,馬上ctrl+c,會觸發(fā)run_until_complete的執(zhí)行異常 KeyBorardInterrupt。然后通過循環(huán)asyncio.Task取消future。
True表示cannel成功,loop stop之后還需要再次開啟事件循環(huán),最后在close,不然還會拋出異常
循環(huán)task,逐個cancel是一種方案,可是正如上面我們把task的列表封裝在main函數中,main函數外進行事件循環(huán)的調用。這個時候,main相當于最外出的一個task,那么處理包裝的main函數即可。
協(xié)程中生產-消費模型設計
通過上面的動態(tài)添加協(xié)程的思想,我們可以設計一個生產-消費的模型,至于中間件(管道)是什么無所謂,下面以內置隊列和redis隊列來舉例說明。
提示:若想主線程退出時,子線程也隨之退出,需要將子線程設置為守護線程,函數 setDaemon(True)
import asyncio
from threading import Thread
from collections import deque
import random
import time
def start_thread_loop(loop):
asyncio.set_event_loop(loop)
loop.run_forever()
def consumer():
while True:
if dq:
msg = dq.pop()
if msg:
asyncio.run_coroutine_threadsafe(thread_example('Zarten'+ msg), new_loop)
async def thread_example(name):
print('正在執(zhí)行name:', name)
await asyncio.sleep(2)
return '返回結果:' + name
dq = deque()
new_loop = asyncio.new_event_loop()
loop_thread = Thread(target= start_thread_loop, args=(new_loop,))
loop_thread.setDaemon(True)
loop_thread.start()
consumer_thread = Thread(target= consumer)
consumer_thread.setDaemon(True)
consumer_thread.start()
while True:
i = random.randint(1, 10)
dq.appendleft(str(i))
time.sleep(2)
redis隊列模型
生產者代碼:
import redis
conn_pool = redis.ConnectionPool(host='127.0.0.1')
redis_conn = redis.Redis(connection_pool=conn_pool)
redis_conn.lpush('coro_test', '1')
redis_conn.lpush('coro_test', '2')
redis_conn.lpush('coro_test', '3')
redis_conn.lpush('coro_test', '4')
消費者代碼:
import asyncio
from threading import Thread
import redis
def get_redis():
conn_pool = redis.ConnectionPool(host= '127.0.0.1')
return redis.Redis(connection_pool= conn_pool)
def start_thread_loop(loop):
asyncio.set_event_loop(loop)
loop.run_forever()
async def thread_example(name):
print('正在執(zhí)行name:', name)
await asyncio.sleep(2)
return '返回結果:' + name
redis_conn = get_redis()
new_loop = asyncio.new_event_loop()
loop_thread = Thread(target= start_thread_loop, args=(new_loop,))
loop_thread.setDaemon(True)
loop_thread.start()
#循環(huán)接收redis消息并動態(tài)加入協(xié)程
while True:
msg = redis_conn.rpop('coro_test')
if msg:
asyncio.run_coroutine_threadsafe(thread_example('Zarten' + bytes.decode(msg, 'utf-8')), new_loop)
asyncio在aiohttp中的應用
aiohttp是一個異步庫,分為客戶端和服務端,下面只是簡單對客戶端做個介紹以及一個經常遇到的異常情況。
aiohttp客戶端為異步網絡請求庫
import asyncio
import aiohttp
count = 0
async def get_http(url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as res:
global count
count += 1
print(count, res.status)
def main():
loop = asyncio.get_event_loop()
url = 'https://www.baidu.com/s?ie=utf-8&f=8&rsv_bp=1&ch=&tn=baiduerr&bar=&wd={0}'
tasks = [get_http(url.format(i)) for i in range(10)]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()
if __name__ == '__main__':
main()
aiohttp并發(fā)量太大的異常解決方案
在使用aiohttp客戶端進行大量并發(fā)請求時,程序會拋出 ValueError: too many file descriptors in select() 的錯誤。
異常代碼示例
說明:測試機器為windows系統(tǒng)
import asyncio
import aiohttp
count = 0
async def get_http(url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as res:
global count
count += 1
print(count, res.status)
def main():
loop = asyncio.get_event_loop()
url = 'https://www.baidu.com/s?ie=utf-8&f=8&rsv_bp=1&ch=&tn=baiduerr&bar=&wd={0}'
tasks = [get_http(url.format(i)) for i in range(600)]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()
if __name__ == '__main__':
main()
原因分析:使用aiohttp時,python內部會使用select(),操作系統(tǒng)對文件描述符最大數量有限制,linux為1024個,windows為509個。
解決方案:
最常見的解決方案是:限制并發(fā)數量(一般500),若并發(fā)的量不大可不作限制。其他方案這里不做介紹,如windows下使用loop = asyncio.ProactorEventLoop() 以及使用回調方式等
限制并發(fā)數量方法
提示:此方法也可用來作為異步爬蟲的限速方法(反反爬)
使用semaphore = asyncio.Semaphore(500) 以及在協(xié)程中使用 async with semaphore: 操作
具體代碼如下:
import asyncio
import aiohttp
async def get_http(url):
async with semaphore:
async with aiohttp.ClientSession() as session:
async with session.get(url) as res:
global count
count += 1
print(count, res.status)
if __name__ == '__main__':
count = 0
semaphore = asyncio.Semaphore(500)
loop = asyncio.get_event_loop()
url = 'https://www.baidu.com/s?ie=utf-8&f=8&rsv_bp=1&ch=&tn=baiduerr&bar=&wd={0}'
tasks = [get_http(url.format(i)) for i in range(600)]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()
在線程或進程池中執(zhí)行代碼
在《流暢的python》中有這樣一段話。
函數(例如io讀寫,requests網絡請求)阻塞了客戶代碼與asycio事件循環(huán)的唯一線程,因此在執(zhí)行調用時,整個應用程序都會凍結。這個問題的解決方法是,使用事件循環(huán)對象的 run_in_executor方法。
asyncio的事件循環(huán)在背后維護著一個ThreadPoolExecutor對象,我們可以調用run_in_executor方法,把可調用對象發(fā)給它執(zhí)行。
import asyncio
from time import sleep, strftime
from concurrent import futures
executor = futures.ThreadPoolExecutor(max_workers=5)
async def blocked_sleep(name, t):
print(strftime('[%H:%M:%S]'), end=' ')
print('sleep {} is running {}s'.format(name, t))
loop = asyncio.get_event_loop()
await loop.run_in_executor(executor, sleep, t)
print(strftime('[%H:%M:%S]'), end=' ')
print('sleep {} is end'.format(name))
return t
async def main():
future = (blocked_sleep(i, i) for i in range(1, 6))
fs = asyncio.gather(*future)
return await fs
loop = asyncio.get_event_loop()
results = loop.run_until_complete(main())
print('results: {}'.format(results))
在同一個線程里,兩個 event loop 無法同時 run,但這不能阻止您用兩個線程分別跑兩個 event loop,
其次再說 ThreadPoolExecutor。您也可以看到,它根本不是 asyncio 庫的東西。當您創(chuàng)建一個 ThreadPoolExecutor 對象時,您實際上是創(chuàng)建了一個線程池。
僅此而已,與 asyncio、event loop 并無瓜葛。而當您明確使用一個 event loop 的 run_in_executor() 方法時,其實底層做的只有兩件事:
1,用線程池執(zhí)行給定函數,與 asyncio 毫無關系;
2,給線程池執(zhí)行結果增加一個回調,該回調會在 event loop 的下一次循環(huán)中保存執(zhí)行結果。
所以 run_in_executor() 只是將傳統(tǒng)的線程池結果拉回到給定 event loop 中,以便進一步處理而已,不存在誰共享誰的關系,指定誰是誰。您可以嘗試一下,在多個線程中跑多個 event loop,然后都向同一個線程池扔任務,然后返回結果:
import asyncio
import threading
import time
from concurrent.futures.thread import ThreadPoolExecutor
e = ThreadPoolExecutor()
def worker(index):
print(index, 'before:', time.strftime('%X'))
time.sleep(1)
print(index, 'after:', time.strftime('%X'))
return index
def main(index):
loop = asyncio.new_event_loop()
res = loop.run_until_complete(loop.run_in_executor(e, worker, index))
print('Thread', index, 'got result', res)
threads = []
for i in range(5):
t = threading.Thread(target=main, args=(i,))
t.start()
threads.append(t)
for t in threads:
t.join()
不同于上面的方法,這里是把阻塞的方法放到新的線程里跑。
總結
以上為個人經驗,希望能給大家一個參考,也希望大家多多支持腳本之家。
相關文章
Python實現清理微信僵尸粉功能示例【基于itchat模塊】
這篇文章主要介紹了Python實現清理微信僵尸粉功能,結合實例形式分析了Python使用itchat模塊刪除微信僵尸粉的相關原理、操作技巧與注意事項,需要的朋友可以參考下2020-05-05
Python Pygame實戰(zhàn)之飛機大戰(zhàn)的實現
飛機大戰(zhàn)想必是很多人童年時期的經典游戲,這篇文章主要給大家介紹了關于如何利用python中的Pygame模塊寫一個簡單的飛機大戰(zhàn)小游戲的相關資料,需要的朋友可以參考下2022-03-03
django xadmin action兼容自定義model權限教程
這篇文章主要介紹了django xadmin action兼容自定義model權限教程,具有很好的參考價值,希望對大家有所幫助。一起跟隨小編過來看看吧2020-03-03

