Python asyncio 异步编程实战：从回调地狱到优雅并发

1
import asyncio
2

3
async def say_hello(name: str, delay: int):
4
    """一个协程函数"""
5
    print(f"开始处理 {name}")
6
    await asyncio.sleep(delay)  # 模拟 I/O 等待，主动让出控制权
7
    print(f"{name} 处理完成")
8
    return f"Hello, {name}!"
9

10
async def main():
11
    # 并发运行三个协程
12
    results = await asyncio.gather(
13
        say_hello("Alice", 2),
14
        say_hello("Bob", 1),
15
        say_hello("Charlie", 3),
16
    )
17
    print(results)
18

19
asyncio.run(main())

1
import asyncio
2
import aiohttp
3
import time
4

5
async def fetch_url(session: aiohttp.ClientSession, url: str) -> dict:
6
    """异步请求单个 URL"""
7
    try:
8
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
9
            return {
10
                "url": url,
11
                "status": resp.status,
12
                "size": len(await resp.text()),
13
            }
14
    except Exception as e:
15
        return {"url": url, "error": str(e)}
16

17
async def batch_fetch(urls: list[str], concurrency: int = 20) -> list[dict]:
18
    """带并发控制的批量请求"""
19
    semaphore = asyncio.Semaphore(concurrency)
20

21
    async def limited_fetch(session, url):
22
        async with semaphore:  # 限制同时并发数
23
            return await fetch_url(session, url)
24

25
    async with aiohttp.ClientSession() as session:
26
        tasks = [limited_fetch(session, url) for url in urls]
27
        return await asyncio.gather(*tasks)
28

29
async def main():
30
    urls = [
31
        "https://httpbin.org/get",
32
        "https://httpbin.org/ip",
33
        "https://httpbin.org/headers",
34
        "https://httpbin.org/user-agent",
35
    ] * 25  # 100 个请求
36

37
    start = time.time()
38
    results = await batch_fetch(urls, concurrency=20)
39
    elapsed = time.time() - start
40

41
    success = sum(1 for r in results if "status" in r)
42
    failed = sum(1 for r in results if "error" in r)
43

44
    print(f"总请求: {len(results)}")
45
    print(f"成功: {success}, 失败: {failed}")
46
    print(f"总耗时: {elapsed:.2f}s")
47
    print(f"平均每个请求: {elapsed/len(results)*1000:.0f}ms")
48

49
asyncio.run(main())

1
import asyncio
2
import random
3

4
async def producer(queue: asyncio.Queue, name: str, count: int):
5
    """生产者：生成任务"""
6
    for i in range(count):
7
        item = f"{name}_item_{i}"
8
        await queue.put(item)
9
        print(f"[生产者 {name}] 放入: {item}")
10
        await asyncio.sleep(random.uniform(0.1, 0.5))
11

12
    print(f"[生产者 {name}] 完成")
13

14
async def consumer(queue: asyncio.Queue, name: str):
15
    """消费者：处理任务"""
16
    while True:
17
        item = await queue.get()
18
        print(f"[消费者 {name}] 处理: {item}")
19
        await asyncio.sleep(random.uniform(0.5, 1.5))  # 模拟处理耗时
20
        queue.task_done()
21
        print(f"[消费者 {name}] 完成: {item}")
22

23
async def main():
24
    queue = asyncio.Queue(maxsize=10)  # 有界队列，防止内存爆炸
25

26
    # 2 个生产者，3 个消费者
27
    producers = [
28
        asyncio.create_task(producer(queue, "P1", 5)),
29
        asyncio.create_task(producer(queue, "P2", 5)),
30
    ]
31
    consumers = [
32
        asyncio.create_task(consumer(queue, f"C{i}"))
33
        for i in range(3)
34
    ]
35

36
    # 等待所有生产者完成
37
    await asyncio.gather(*producers)
38

39
    # 等待队列清空
40
    await queue.join()
41

42
    # 取消消费者（它们是无限循环）
43
    for c in consumers:
44
        c.cancel()
45

46
    print("所有任务处理完成！")
47

48
asyncio.run(main())

1
from fastapi import FastAPI
2
import aiohttp
3
import asyncio
4

5
app = FastAPI()
6

7
async def fetch_weather(city: str) -> dict:
8
    """异步获取天气数据"""
9
    async with aiohttp.ClientSession() as session:
10
        url = f"https://wttr.in/{city}?format=j1"
11
        async with session.get(url) as resp:
12
            return await resp.json()
13

14
@app.get("/compare-weather")
15
async def compare_weather(cities: str = "Beijing,Shanghai,Guangzhou"):
16
    """并发获取多个城市天气"""
17
    city_list = cities.split(",")
18

19
    # 并发请求所有城市的天气
20
    tasks = [fetch_weather(city.strip()) for city in city_list]
21
    results = await asyncio.gather(*tasks, return_exceptions=True)
22

23
    return {
24
        city: data if not isinstance(data, Exception) else {"error": str(data)}
25
        for city, data in zip(city_list, results)
26
    }