Python 中這十個異步 IO 技巧，asyncio_aiohttp 進階！
Python中的異步IO是近年來非常熱門的一個話題，特別是在處理網絡請求、爬蟲、實時數據處理等場景中，異步IO可以顯著提升程序的性能。今天我們就來聊聊Python中異步IO的10個實用技巧，涵蓋asyncio和aiohttp的進階用法。
我們先從最基礎的async/await語法開始，它是Python異步編程的核心。
1. 使用async/await定義協程函數
在Python中，我們可以使用async def來定義一個協程函數，然后使用await來調用其他協程。
import asyncio
async def say_hello():
print("Hello")
await asyncio.sleep(1) # 模擬異步操作
print("World")
asyncio.run(say_hello())這段代碼定義了一個協程函數say_hello，它在執行時會先打印“Hello”，然后等待1秒，再打印“World”。asyncio.run()是Python 3.7之后引入的，用于啟動主函數。
2. 使用asyncio.gather并發執行多個協程
如果我們有多個協程，可以使用asyncio.gather()來并發執行它們。
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())這里，我們同時啟動了三個任務，它們會并發執行，而不是順序執行。
3. 使用asyncio.create_task創建任務
asyncio.create_task()用于創建一個任務對象，可以在主函數中并發執行。
import asyncio
async def say_hello():
print("Hello")
await asyncio.sleep(1)
print("World")
async def main():
task = asyncio.create_task(say_hello()) # 創建任務
print("Main function")
await task # 等待任務完成
asyncio.run(main())這個例子中，say_hello()協程被創建為一個任務，主函數繼續執行，而任務在后臺運行。
4. 使用asyncio.sleep模擬異步IO操作
asyncio.sleep()用于模擬異步IO操作，比如網絡請求、文件讀寫等。
import asyncio
async def fetch_data():
print("Fetching data...")
await asyncio.sleep(2) # 模擬網絡請求耗時
print("Data fetched")
asyncio.run(fetch_data())5. 使用aiohttp進行異步HTTP請求
aiohttp是一個非常強大的異步HTTP客戶端/服務器庫，適合用于爬蟲、API調用等場景。
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發起一個GET請求，并獲取響應內容。
6. 使用aiohttp并發請求多個URL
我們可以結合asyncio.gather()，同時發起多個HTTP請求。
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. 設置超時時間
在使用aiohttp時，可以為請求設置超時時間，避免長時間等待。
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進行異步任務隊列管理
在處理大量異步任務時，可以使用asyncio.Queue來管理任務隊列。
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進行異步鎖控制
在多協程環境下，使用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調用子進程
asyncio也可以用于異步地執行系統命令。
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())實戰案例：使用aiohttp爬取多個網頁內容
場景： 我們需要爬取多個網頁的內容，并將結果保存到本地文件。
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())這個實戰案例展示了如何使用aiohttp并發抓取網頁內容，并保存到本地。
總結
本文介紹了Python異步IO的10個實用技巧，從async/await語法、asyncio.gather()、asyncio.create_task()，到使用aiohttp進行異步網絡請求、任務隊列管理、超時控制等。通過這些技巧，我們可以顯著提升Python程序的性能和效率。這些知識在實際開發中非常有用，特別是在處理網絡請求、并發任務、數據爬取等場景中。
