Python 中這十個(gè)異步 IO 技巧，asyncio_aiohttp 進(jìn)階！
Python中的異步IO是近年來非常熱門的一個(gè)話題，特別是在處理網(wǎng)絡(luò)請(qǐng)求、爬蟲、實(shí)時(shí)數(shù)據(jù)處理等場(chǎng)景中，異步IO可以顯著提升程序的性能。今天我們就來聊聊Python中異步IO的10個(gè)實(shí)用技巧，涵蓋asyncio和aiohttp的進(jìn)階用法。
我們先從最基礎(chǔ)的async/await語法開始，它是Python異步編程的核心。
1. 使用async/await定義協(xié)程函數(shù)
在Python中，我們可以使用async def來定義一個(gè)協(xié)程函數(shù)，然后使用await來調(diào)用其他協(xié)程。
import asyncio
async def say_hello():
print("Hello")
await asyncio.sleep(1) # 模擬異步操作
print("World")
asyncio.run(say_hello())這段代碼定義了一個(gè)協(xié)程函數(shù)say_hello，它在執(zhí)行時(shí)會(huì)先打印“Hello”，然后等待1秒，再打印“World”。asyncio.run()是Python 3.7之后引入的，用于啟動(dòng)主函數(shù)。
2. 使用asyncio.gather并發(fā)執(zhí)行多個(gè)協(xié)程
如果我們有多個(gè)協(xié)程，可以使用asyncio.gather()來并發(fā)執(zhí)行它們。
import asyncio
asyncdef task(name):
print(f"Task {name} started")
await asyncio.sleep(1)
print(f"Task {name} finished")
asyncdef main():
await asyncio.gather(
task("A"),
task("B"),
task("C")
)
asyncio.run(main())這里，我們同時(shí)啟動(dòng)了三個(gè)任務(wù)，它們會(huì)并發(fā)執(zhí)行，而不是順序執(zhí)行。
3. 使用asyncio.create_task創(chuàng)建任務(wù)
asyncio.create_task()用于創(chuàng)建一個(gè)任務(wù)對(duì)象，可以在主函數(shù)中并發(fā)執(zhí)行。
import asyncio
async def say_hello():
print("Hello")
await asyncio.sleep(1)
print("World")
async def main():
task = asyncio.create_task(say_hello()) # 創(chuàng)建任務(wù)
print("Main function")
await task # 等待任務(wù)完成
asyncio.run(main())這個(gè)例子中，say_hello()協(xié)程被創(chuàng)建為一個(gè)任務(wù)，主函數(shù)繼續(xù)執(zhí)行，而任務(wù)在后臺(tái)運(yùn)行。
4. 使用asyncio.sleep模擬異步IO操作
asyncio.sleep()用于模擬異步IO操作，比如網(wǎng)絡(luò)請(qǐng)求、文件讀寫等。
import asyncio
async def fetch_data():
print("Fetching data...")
await asyncio.sleep(2) # 模擬網(wǎng)絡(luò)請(qǐng)求耗時(shí)
print("Data fetched")
asyncio.run(fetch_data())5. 使用aiohttp進(jìn)行異步HTTP請(qǐng)求
aiohttp是一個(gè)非常強(qiáng)大的異步HTTP客戶端/服務(wù)器庫，適合用于爬蟲、API調(diào)用等場(chǎng)景。
import aiohttp
import asyncio
asyncdef fetch(session, url):
asyncwith session.get(url) as response:
returnawait response.text()
asyncdef main():
asyncwith aiohttp.ClientSession() as session:
html = await fetch(session, 'https://httpbin.org/get')
print(html)
asyncio.run(main())這段代碼使用aiohttp發(fā)起一個(gè)GET請(qǐng)求，并獲取響應(yīng)內(nèi)容。
6. 使用aiohttp并發(fā)請(qǐng)求多個(gè)URL
我們可以結(jié)合asyncio.gather()，同時(shí)發(fā)起多個(gè)HTTP請(qǐng)求。
import aiohttp
import asyncio
asyncdef fetch(session, url):
asyncwith session.get(url) as response:
returnawait response.text()
asyncdef main():
urls = ['https://httpbin.org/get', 'https://httpbin.org/user-agent']
asyncwith aiohttp.ClientSession() as session:
tasks = [fetch(session, url) for url in urls]
results = await asyncio.gather(*tasks)
for result in results:
print(result)
asyncio.run(main())7. 設(shè)置超時(shí)時(shí)間
在使用aiohttp時(shí)，可以為請(qǐng)求設(shè)置超時(shí)時(shí)間，避免長時(shí)間等待。
async def fetch(session, url):
try:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as response:
return await response.text()
except Exception as e:
print(f"Error: {e}")
return None8. 使用asyncio.Queue進(jìn)行異步任務(wù)隊(duì)列管理
在處理大量異步任務(wù)時(shí)，可以使用asyncio.Queue來管理任務(wù)隊(duì)列。
import asyncio
asyncdef worker(name, queue):
whileTrue:
item = await queue.get()
print(f"Worker {name} processing {item}")
await asyncio.sleep(1)
queue.task_done()
asyncdef main():
queue = asyncio.Queue()
for i in range(10):
queue.put_nowait(f"item_{i}")
workers = [worker(f"Worker_{i}", queue) for i in range(3)]
await asyncio.gather(*workers)
await queue.join()
asyncio.run(main())9. 使用asyncio.Lock進(jìn)行異步鎖控制
在多協(xié)程環(huán)境下，使用asyncio.Lock可以避免資源競爭。
import asyncio
lock = asyncio.Lock()
asyncdef print_counter(name):
asyncwith lock:
for i in range(3):
print(f"{name}: {i}")
await asyncio.sleep(0.5)
asyncdef main():
await asyncio.gather(
print_counter("A"),
print_counter("B")
)
asyncio.run(main())10. 使用asyncio.subprocess調(diào)用子進(jìn)程
asyncio也可以用于異步地執(zhí)行系統(tǒng)命令。
import asyncio
async def run_cmd():
proc = await asyncio.create_subprocess_shell(
'ping 127.0.0.1 -n 4',
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await proc.communicate()
print(f"STDOUT: {stdout.decode()}")
print(f"STDERR: {stderr.decode()}")
asyncio.run(run_cmd())實(shí)戰(zhàn)案例：使用aiohttp爬取多個(gè)網(wǎng)頁內(nèi)容
場(chǎng)景： 我們需要爬取多個(gè)網(wǎng)頁的內(nèi)容，并將結(jié)果保存到本地文件。
import aiohttp
import asyncio
import os
asyncdef fetch(session, url, filename):
asyncwith session.get(url) as response:
content = await response.text()
with open(filename, 'w', encoding='utf-8') as f:
f.write(content)
print(f"Saved {filename}")
asyncdef main():
urls = [
'https://httpbin.org/get',
'https://httpbin.org/user-agent'
]
os.makedirs('output', exist_ok=True)
asyncwith aiohttp.ClientSession() as session:
tasks = [
fetch(session, url, f'output/{i}.html') for i, url in enumerate(urls)
]
await asyncio.gather(*tasks)
asyncio.run(main())這個(gè)實(shí)戰(zhàn)案例展示了如何使用aiohttp并發(fā)抓取網(wǎng)頁內(nèi)容，并保存到本地。
總結(jié)
本文介紹了Python異步IO的10個(gè)實(shí)用技巧，從async/await語法、asyncio.gather()、asyncio.create_task()，到使用aiohttp進(jìn)行異步網(wǎng)絡(luò)請(qǐng)求、任務(wù)隊(duì)列管理、超時(shí)控制等。通過這些技巧，我們可以顯著提升Python程序的性能和效率。這些知識(shí)在實(shí)際開發(fā)中非常有用，特別是在處理網(wǎng)絡(luò)請(qǐng)求、并發(fā)任務(wù)、數(shù)據(jù)爬取等場(chǎng)景中。
