🐍 第29课：异步编程

—— asyncio：单线程并发的终极武器

📌 本课目标

1️⃣ 协程基础

import asyncio

# 定义协程
async def hello(name):
    print(f"开始: {name}")
    await asyncio.sleep(1)  # 非阻塞等待
    print(f"完成: {name}")
    return f"result_{name}"

# 运行协程
async def main():
    # 串行执行（总耗时约2秒）
    result1 = await hello("A")
    result2 = await hello("B")
    print(f"串行: {result1}, {result2}")

asyncio.run(main())

# 并发执行（总耗时约1秒）
async def main_concurrent():
    results = await asyncio.gather(
        hello("A"),
        hello("B"),
        hello("C"),
    )
    print(f"并发: {results}")

asyncio.run(main_concurrent())

2️⃣ 任务与调度

import asyncio

# 创建 Task
async def task_example():
    async def work(name, seconds):
        print(f"{name} 开始")
        await asyncio.sleep(seconds)
        print(f"{name} 完成（耗时{seconds}s）")
        return f"{name}_result"
    
    # 方式一：create_task
    task1 = asyncio.create_task(work("T1", 2))
    task2 = asyncio.create_task(work("T2", 1))
    
    # 等待所有完成
    results = await asyncio.gather(task1, task2)
    print(f"结果: {results}")
    
    # 方式二：as_completed（按完成顺序）
    tasks = [asyncio.create_task(work(f"W{i}", i*0.5)) for i in range(1, 4)]
    for coro in asyncio.as_completed(tasks):
        result = await coro
        print(f"  最早完成: {result}")
    
    # 方式三：wait（更灵活）
    tasks = [asyncio.create_task(work(f"J{i}", i)) for i in range(1, 4)]
    done, pending = await asyncio.wait(
        tasks,
        timeout=2.0,           # 最多等2秒
        return_when=asyncio.FIRST_COMPLETED,
    )
    print(f"已完成: {len(done)}, 未完成: {len(pending)}")
    for t in pending:
        t.cancel()  # 取消未完成的任务

asyncio.run(task_example())

3️⃣ aiohttp 异步 HTTP

import asyncio
import aiohttp

async def fetch_url(session, url):
    """异步获取 URL"""
    async with session.get(url) as resp:
        status = resp.status
        text = await resp.text()
        return {"url": url, "status": status, "length": len(text)}

async def fetch_many(urls, max_concurrent=5):
    """并发获取多个 URL"""
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def limited_fetch(session, url):
        async with semaphore:
            return await fetch_url(session, url)
    
    async with aiohttp.ClientSession() as session:
        tasks = [limited_fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
    
    return results

# 使用
async def main():
    urls = [
        "https://httpbin.org/get",
        "https://httpbin.org/ip",
        "https://httpbin.org/headers",
    ]
    results = await fetch_many(urls, max_concurrent=3)
    for r in results:
        if isinstance(r, Exception):
            print(f"  ❌ 异常: {r}")
        else:
            print(f"  ✅ {r['url']} → {r['status']} ({r['length']} bytes)")

# asyncio.run(main())  # 取消注释运行

4️⃣ 异步文件与定时任务

import asyncio
import aiofiles  # pip install aiofiles
import json

async def async_write_json(filepath, data):
    """异步写入 JSON"""
    async with aiofiles.open(filepath, "w") as f:
        await f.write(json.dumps(data, ensure_ascii=False, indent=2))

async def async_read_json(filepath):
    """异步读取 JSON"""
    async with aiofiles.open(filepath) as f:
        content = await f.read()
        return json.loads(content)

# 异步定时任务
async def periodic_task(interval, task_func, *args):
    """定时执行协程任务"""
    while True:
        try:
            await task_func(*args)
        except Exception as e:
            print(f"定时任务异常: {e}")
        await asyncio.sleep(interval)

async def heartbeat():
    """心跳检查"""
    import time
    print(f"  💓 心跳 {time.strftime('%H:%M:%S')}")

async def main():
    # 运行心跳5秒
    task = asyncio.create_task(periodic_task(1, heartbeat))
    await asyncio.sleep(5)
    task.cancel()

# asyncio.run(main())

5️⃣ 验证脚本

#!/usr/bin/env python3
"""第29课 异步编程验证"""
import asyncio
import time

async def test_basic_coroutine():
    async def add(a, b):
        await asyncio.sleep(0.01)
        return a + b
    
    result = await add(3, 4)
    assert result == 7
    print("✅ 基础协程测试通过")

async def test_gather():
    async def work(n):
        await asyncio.sleep(0.01)
        return n * 2
    
    results = await asyncio.gather(work(1), work(2), work(3))
    assert results == [2, 4, 6]
    print("✅ gather并发测试通过")

async def test_semaphore():
    sem = asyncio.Semaphore(2)
    active = 0
    max_active = 0
    
    async def limited_work():
        nonlocal active, max_active
        async with sem:
            active += 1
            max_active = max(max_active, active)
            await asyncio.sleep(0.01)
            active -= 1
    
    tasks = [limited_work() for _ in range(10)]
    await asyncio.gather(*tasks)
    assert max_active <= 2
    print(f"✅ Semaphore测试通过 (最大并发: {max_active})")

async def test_timeout():
    async def slow_task():
        await asyncio.sleep(10)
    
    try:
        await asyncio.wait_for(slow_task(), timeout=0.1)
        assert False, "Should have timed out"
    except asyncio.TimeoutError:
        print("✅ 超时控制测试通过")

async def test_task_cancel():
    async def long_task():
        try:
            await asyncio.sleep(100)
        except asyncio.CancelledError:
            return "cancelled"
    
    task = asyncio.create_task(long_task())
    await asyncio.sleep(0.01)
    task.cancel()
    result = await task
    assert result == "cancelled"
    print("✅ 任务取消测试通过")

async def run_all():
    await test_basic_coroutine()
    await test_gather()
    await test_semaphore()
    await test_timeout()
    await test_task_cancel()
    print("\n🎉 第29课全部验证通过！")

if __name__ == "__main__":
    asyncio.run(run_all())

6️⃣ 异步上下文管理器

import asyncio
import time

class AsyncTimer:
    """异步计时器"""
    async def __aenter__(self):
        self.start = time.time()
        return self
    async def __aexit__(self, *args):
        self.elapsed = time.time() - self.start
        print(f"耗时: {self.elapsed:.3f}s")

async def demo_timer():
    async with AsyncTimer():
        await asyncio.sleep(0.5)
        await asyncio.sleep(0.3)

# asyncio.run(demo_timer())

7️⃣ 异步生产者-消费者

import asyncio
import random

class AsyncProducerConsumer:
    """异步生产者-消费者模式"""
    
    def __init__(self, queue_size=10, num_producers=2, num_consumers=3):
        self.queue = asyncio.Queue(maxsize=queue_size)
        self.num_producers = num_producers
        self.num_consumers = num_consumers
        self.produced = 0
        self.consumed = 0
    
    async def producer(self, pid):
        for i in range(5):
            item = f"P{pid}-item{i}"
            await self.queue.put(item)
            self.produced += 1
            await asyncio.sleep(random.uniform(0.01, 0.05))
    
    async def consumer(self, cid):
        while True:
            item = await self.queue.get()
            self.consumed += 1
            await asyncio.sleep(random.uniform(0.02, 0.08))
            self.queue.task_done()
    
    async def run(self):
        consumers = [asyncio.create_task(self.consumer(i)) for i in range(self.num_consumers)]
        producers = [asyncio.create_task(self.producer(i)) for i in range(self.num_producers)]
        await asyncio.gather(*producers)
        await self.queue.join()
        for c in consumers:
            c.cancel()
        print(f"统计: 生产{self.produced} 消费{self.consumed}")

# asyncio.run(AsyncProducerConsumer().run())

8️⃣ asyncio 常见陷阱

import asyncio
import time

# 陷阱1：在 async 函数中使用 time.sleep（阻塞整个事件循环）
async def bad_sleep():
    time.sleep(1)  # ❌ 阻塞所有协程

async def good_sleep():
    await asyncio.sleep(1)  # ✅ 非阻塞

# 陷阱2：忘记 await
async def forgot_await():
    coro = good_sleep()  # 创建协程但没执行
    # 忘记 await coro  # ❌ 协程没执行
    await coro  # ✅ 正确

# 陷阱3：在 async 中用 requests（阻塞）
# import requests  # ❌ 同步 HTTP 库，阻塞事件循环
# import aiohttp   # ✅ 异步 HTTP 库

# 陷阱4：gather 不处理异常
async def may_fail():
    raise ValueError("oops")

async def safe_gather():
    results = await asyncio.gather(
        may_fail(), may_fail(),
        return_exceptions=True  # ✅ 异常作为返回值而非抛出
    )
    for r in results:
        if isinstance(r, Exception):
            print(f"任务失败: {r}")
        else:
            print(f"任务成功: {r}")

# asyncio.run(safe_gather())

9️⃣ 异步迭代器

import asyncio

class AsyncRange:
    """异步迭代器示例"""
    def __init__(self, start, end, delay=0.1):
        self.current = start
        self.end = end
        self.delay = delay
    
    def __aiter__(self):
        return self
    
    async def __anext__(self):
        if self.current >= self.end:
            raise StopAsyncIteration
        value = self.current
        self.current += 1
        await asyncio.sleep(self.delay)
        return value

async def demo_async_iterator():
    total = 0
    async for num in AsyncRange(1, 6, delay=0.05):
        print(f"  处理: {num}")
        total += num
    print(f"  总和: {total}")

# asyncio.run(demo_async_iterator())

# 异步生成器
async def async_fibonacci(n):
    a, b = 0, 1
    for _ in range(n):
        yield a
        await asyncio.sleep(0.01)
        a, b = b, a + b

async def demo_generator():
    fibs = [n async for n in async_fibonacci(10)]
    print(f"斐波那契: {fibs}")

# asyncio.run(demo_generator())

🔟 异步生态工具

"""异步生态常用库：
- aiohttp    → 异步 HTTP 客户端/服务端
- aiofiles   → 异步文件操作
- aiomysql   → 异步 MySQL
- aioredis   → 异步 Redis
- aiosqlite  → 异步 SQLite
- asyncpg    → 异步 PostgreSQL
- httpx      → 同步+异步 HTTP（推荐替代 requests）

选择建议：
- 新项目首选 httpx（同时支持同步/异步）
- 数据库用 SQLAlchemy 2.0（原生 async 支持）
- 文件操作：小文件同步即可，大文件用 aiofiles
"""

🔟 性能对比：同步 vs 异步

import asyncio
import time
import concurrent.futures

# 模拟100个HTTP请求，每个耗时0.1秒

def sync_fetch_all(n=100):
    """同步串行"""
    start = time.time()
    for i in range(n):
        time.sleep(0.01)  # 模拟请求
    return time.time() - start

async def async_fetch_all(n=100):
    """异步并发"""
    start = time.time()
    async def fetch(i):
        await asyncio.sleep(0.01)  # 非阻塞等待
    await asyncio.gather(*[fetch(i) for i in range(n)])
    return time.time() - start

def threaded_fetch_all(n=100, workers=20):
    """多线程并发"""
    start = time.time()
    with concurrent.futures.ThreadPoolExecutor(workers) as pool:
        list(pool.map(lambda _: time.sleep(0.01), range(n)))
    return time.time() - start

# 结果对比（100个请求，每个0.01秒）
# 串行: ~1.0s
# 多线程(20): ~0.05s
# 异步: ~0.01s（最快！）

🔑 本课要点