我們一起聊聊Python協(xié)程和異步編程
作者：華安9527
Python中的異步編程主要依賴于`asyncio`模塊。`asyncio`提供了一套用于編寫異步代碼的工具和框架，包括協(xié)程、事件循環(huán)和異步IO操作等。
協(xié)程和異步編程是Python中處理并發(fā)和異步任務的重要概念。協(xié)程是一種輕量級的并發(fā)編程方式，它允許程序在執(zhí)行過程中暫停和恢復，以便處理其他任務。異步編程模型則是基于協(xié)程的一種編程風格，它通過使用非阻塞的異步IO操作來提高程序的并發(fā)性能。
Python中的異步編程主要依賴于`asyncio`模塊。`asyncio`提供了一套用于編寫異步代碼的工具和框架，包括協(xié)程、事件循環(huán)和異步IO操作等。
代碼示例：
1. 使用`async`和`await`定義協(xié)程函數：
import asyncio
async def my_coroutine():
await asyncio.sleep(1)
print("Coroutine executed")
asyncio.run(my_coroutine())2. 使用`asyncio.create_task()`并發(fā)運行多個協(xié)程：
import asyncio
async def coroutine1():
await asyncio.sleep(1)
print("Coroutine 1 executed")
async def coroutine2():
await asyncio.sleep(2)
print("Coroutine 2 executed")
async def main():
task1 = asyncio.create_task(coroutine1())
task2 = asyncio.create_task(coroutine2())
await asyncio.gather(task1, task2)
asyncio.run(main())3. 使用`asyncio.wait()`等待多個協(xié)程完成：
import asyncio
async def coroutine1():
await asyncio.sleep(1)
print("Coroutine 1 executed")
async def coroutine2():
await asyncio.sleep(2)
print("Coroutine 2 executed")
async def main():
tasks = [coroutine1(), coroutine2()]
done, pending = await asyncio.wait(tasks)
for task in done:
print(f"Task {task} completed")
asyncio.run(main())4. 使用`asyncio.Lock()`實現協(xié)程間的互斥訪問：
import asyncio
async def counter(lock):
async with lock:
for _ in range(5):
print("Counting")
await asyncio.sleep(1)
async def main():
lock = asyncio.Lock()
tasks = [counter(lock) for _ in range(3)]
await asyncio.gather(*tasks)
asyncio.run(main())5. 使用`asyncio.Queue()`實現協(xié)程間的消息傳遞：
import asyncio
async def producer(queue):
for i in range(5):
await queue.put(i)
print(f"Produced: {i}")
await asyncio.sleep(1)
async def consumer(queue):
while True:
item = await queue.get()
print(f"Consumed: {item}")
await asyncio.sleep(2)
async def main():
queue = asyncio.Queue()
producer_task = asyncio.create_task(producer(queue))
consumer_task = asyncio.create_task(consumer(queue))
await asyncio.gather(producer_task, consumer_task)
asyncio.run(main())6. 使用`asyncio.TimeoutError`設置協(xié)程的超時：
import asyncio
async def my_coroutine():
await asyncio.sleep(2)
print("Coroutine executed")
async def main():
try:
await asyncio.wait_for(my_coroutine(), timeout=1)
except asyncio.TimeoutError:
print("Coroutine timed out")
asyncio.run(main())7. 使用`asyncio.run_in_executor()`在協(xié)程中執(zhí)行阻塞的同步操作：
import asyncio
def sync_operation():
# 阻塞的同步操作
return "Sync result"
async def main():
loop = asyncio.get_running_loop()
result = await loop.run_in_executor(None, sync_operation)
print(f"Result: {result}")
asyncio.run(main())8. 使用`aiohttp`庫進行異步HTTP請求：
import asyncio
import aiohttp
async def fetch_data(url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.text()
async def main():
url = "https://api.example.com/data"
data = await fetch_data(url)
print(f"Data: {data}")
asyncio.run(main())9. 使用`asyncio.sleep()`模擬異步計時器：
import asyncio
async def timer(duration):
await asyncio.sleep(duration)
print(f"Timer finished after {duration} seconds")
async def main():
tasks = [timer(1), timer(2), timer(3)]
await asyncio.gather(*tasks)
asyncio.run(main())10. 使用`asyncio`實現并發(fā)的文件IO操作：
import asyncio
async def read_file(file):
async with asyncio.open_file(file, "r") as f:
contents = await f.read()
print(f"Read from {file}: {contents}")
async def write_file(file, data):
async with asyncio.open_file(file, "w") as f:
await f.write(data)
print(f"Wrote to {file}")
async def main():
file = "data.txt"
await write_file(file, "Hello, world!")
await read_file(file)
asyncio.run(main())這些場景代碼展示了協(xié)程和異步編程的使用方式。通過使用`asyncio`模塊和相關的工具，我們可以輕松地編寫并發(fā)和異步任務處理的代碼，提高程序的性能和響應能力。
責任編輯：武曉燕
來源：
測試開發(fā)學習交流
