一、什么是協程？

協程（Coroutine）就像可以暫停執行的函數，能夠在執行過程中主動讓出控制權，等準備好后再繼續執行。

生活小例子

想象你在咖啡店排隊：

• 普通函數：必須一直排到取餐（阻塞等待）
• 協程：下單后去旁邊座位等，輪到你再回來取（非阻塞）

二、快速入門

1. 最簡單的協程

import asyncioasync def hello():print("開始")await asyncio.sleep(1)  # 暫停1秒print("結束")asyncio.run(hello())  # 運行協程

2. 并發執行多個協程

async def make_coffee(name, time):print(f"{name}開始制作")await asyncio.sleep(time)print(f"{name}制作完成")async def main():# 同時制作三杯咖啡await asyncio.gather(make_coffee("拿鐵", 2),make_coffee("美式", 1),make_coffee("卡布", 3))asyncio.run(main())

輸出順序：美式 → 拿鐵 → 卡布（總耗時3秒）

三、核心概念

1. 關鍵字解析

關鍵字	作用說明	示例
async	定義協程函數	async def func():
await	暫停等待異步操作	await task()
run()	啟動協程的主入口	asyncio.run(main())

2. 協程 vs 多線程

	協程	多線程
內存占用	約1KB/任務	約8MB/線程
切換速度	100納秒級	1微秒級
適用場景	I/O密集型任務	CPU密集型任務

四、實戰應用

1. 網絡請求并發

import aiohttpasync def fetch(url):async with aiohttp.ClientSession() as session:async with session.get(url) as response:return await response.text()async def main():urls = ["url1", "url2", "url3"]results = await asyncio.gather(*[fetch(url) for url in urls])print(f"獲取到{len(results)}個結果")asyncio.run(main())

2. 生產者-消費者模式

async def producer(queue):for i in range(5):await queue.put(i)print(f"生產產品{i}")await asyncio.sleep(0.5)async def consumer(queue):while True:item = await queue.get()print(f"消費產品{item}")queue.task_done()async def main():queue = asyncio.Queue(3)  # 最大容量3await asyncio.gather(producer(queue),consumer(queue))asyncio.run(main())

五、常見問題

1. 為什么我的協程不執行？

• 忘記使用await調用協程
• 沒有通過asyncio.run()啟動
• 在普通函數中調用協程

2. 如何停止無限循環的協程？

task = asyncio.create_task(infinite_task())
await asyncio.sleep(5)
task.cancel()  # 5秒后取消任務

3. 協程會替代多線程嗎？

• 適合：網絡請求、文件IO、Web服務等I/O密集型場景
• 不適合：科學計算、圖像處理等CPU密集型任務

六、優化

1. 避免阻塞操作：用await asyncio.sleep()代替time.sleep()
2. 限制并發量：

sem = asyncio.Semaphore(10)  # 最多同時10個async def limited_task():async with sem:await heavy_work()

3. 使用結構化并發（Python 3.11+）：

async with asyncio.TaskGroup() as tg:tg.create_task(task1())tg.create_task(task2())

備注

個人水平有限，有問題隨時交流~