⚡ asyncio 异步编程

什么是异步？为什么需要它？

假设你要同时请求 100 个网页。同步方式是一个一个来——等第一个网页返回，再请求第二个。如果每个请求要 1 秒，总共要 100 秒。

异步方式是发完一个请求不等回复，立刻发下一个——100 个请求几乎同时发出，等所有回复一起回来，总共只要 1-2 秒。

💡 提示： 💡 类比：同步 = 你在餐厅点了一道菜，站在厨房门口等做好才点下一道。异步 = 你把 100 道菜全点了，然后坐下来等厨房一起做好端上来。

asyncio 基础

import asyncio

# 定义一个异步函数（协程）
async def fetch_data(url, delay):
    print(f"开始请求 {url}")
    await asyncio.sleep(delay)  # 模拟网络请求（非阻塞等待）
    print(f"完成请求 {url}")
    return f"{url} 的数据"

# 运行协程
async def main():
    # ❌ 同步方式：一个一个来，总耗时 = 所有 delay 之和
    # result1 = await fetch_data("api/users", 2)
    # result2 = await fetch_data("api/orders", 3)
    # 总耗时：5 秒

    # ✅ 异步方式：并发执行，总耗时 = 最长的那个
    results = await asyncio.gather(
        fetch_data("api/users", 2),
        fetch_data("api/orders", 3),
        fetch_data("api/products", 1),
    )
    # 总耗时：3 秒（最长的那个）

    for r in results:
        print(r)

asyncio.run(main())

开始请求 api/users
开始请求 api/orders
开始请求 api/products
完成请求 api/products     ← 1秒后
完成请求 api/users        ← 2秒后
完成请求 api/orders       ← 3秒后（不是 2+3=5秒！）

aiohttp 并发爬虫

真实场景：并发抓取 100 个网页。用 aiohttp 替代 requests：

import asyncio
import aiohttp

async def fetch(session, url):
    """抓取单个网页"""
    try:
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
            text = await resp.text()
            return {"url": url, "status": resp.status, "length": len(text)}
    except Exception as e:
        return {"url": url, "error": str(e)}

async def main():
    urls = [
        "https://httpbin.org/get",
        "https://httpbin.org/delay/1",
        "https://httpbin.org/status/404",
        "https://httpbin.org/html",
    ] * 25  # 模拟 100 个 URL

    # 限制并发数（避免被封 IP）
    semaphore = asyncio.Semaphore(20)

    async def limited_fetch(session, url):
        async with semaphore:
            return await fetch(session, url)

    async with aiohttp.ClientSession() as session:
        # 并发抓取所有 URL（最多 20 个同时进行）
        tasks = [limited_fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)

    # 统计结果
    success = sum(1 for r in results if r.get("status") == 200)
    failed = len(results) - success
    print(f"✅ 成功: {success}, ❌ 失败: {failed}")

asyncio.run(main())

httpx——现代异步 HTTP 客户端

import asyncio
import httpx

async def main():
    # httpx 同时支持同步和异步
    async with httpx.AsyncClient() as client:
        # 并发请求
        tasks = [
            client.get("https://api.github.com/users/torvalds"),
            client.get("https://api.github.com/users/gvanrossum"),
            client.get("https://api.github.com/users/rosalindfranklin"),
        ]
        responses = await asyncio.gather(*tasks)

        for resp in responses:
            data = resp.json()
            print(f"{data['login']}: {data['public_repos']} repos")

asyncio.run(main())

💡 提示： 💡 httpx vs aiohttp：httpx API 更像 requests（学习成本低），同时支持 HTTP/2。aiohttp 更轻量、性能略高。新项目推荐 httpx，已有 aiohttp 代码不需要换。

异步任务管理

import asyncio

async def slow_task():
    await asyncio.sleep(10)
    return "slow done"

async def fast_task():
    await asyncio.sleep(1)
    return "fast done"

async def main():
    # 1. gather：等待所有任务完成
    results = await asyncio.gather(slow_task(), fast_task())

    # 2. wait_for：设置超时（超时抛 TimeoutError）
    try:
        result = await asyncio.wait_for(slow_task(), timeout=3)
    except asyncio.TimeoutError:
        print("任务超时了！")

    # 3. as_completed：按完成顺序处理
    tasks = [fast_task(), slow_task(), fast_task()]
    for coro in asyncio.as_completed(tasks):
        result = await coro
        print(f"完成: {result}")

    # 4. create_task：创建后台任务
    task = asyncio.create_task(slow_task())
    # 可以做其他事情...
    result = await task  # 需要结果时再 await

    # 5. Semaphore：限制并发数
    sem = asyncio.Semaphore(5)  # 最多 5 个并发
    async def limited():
        async with sem:
            return await fast_task()

asyncio.run(main())

异步生成器和上下文管理器

import asyncio

# 异步生成器（用 async for 遍历）
async def async_range(n):
    for i in range(n):
        await asyncio.sleep(0.1)  # 模拟异步操作
        yield i

async def main():
    async for num in async_range(5):
        print(num)  # 0, 1, 2, 3, 4

    # 异步上下文管理器（用 async with）
    class AsyncDB:
        async def __aenter__(self):
            print("连接数据库...")
            await asyncio.sleep(0.5)
            return self

        async def __aexit__(self, *args):
            print("关闭数据库连接...")
            await asyncio.sleep(0.1)

        async def query(self, sql):
            await asyncio.sleep(0.2)
            return [{"id": 1, "name": "test"}]

    async with AsyncDB() as db:
        result = await db.query("SELECT * FROM users")
        print(result)

asyncio.run(main())