wordflow
返回博客

53. 异步 IO：协程与 asyncio

53. 异步 IO：协程与 asyncio

第五十三篇：异步 IO——协程与 asyncio

前言

上一篇我们学习了 Flask Web 框架，但所有的请求都是同步处理的——每个请求占用一个线程，在处理数据库查询、文件 IO、网络请求时会被阻塞。对于高并发场景，这种模型效率很低。异步编程通过单线程下的协作式多任务调度，用极少的资源同时处理成千上万的连接。本文系统讲解 Python 异步编程的核心概念和 asyncio 库的使用方法。

同步 vs 异步 vs 多线程

三种并发模型对比

┌───────────────┬──────────────────┬──────────────────────┐
│    维度        │    同步           │       多线程          │
├───────────────┼──────────────────┼──────────────────────┤
│ 资源消耗       │ 极低（单线程）     │ 较高（OS 线程开销大） │
│ 并发能力       │ 低               │ 中等（~千级）         │
│ CPU 利用率     │ 低（IO 时浪费）   │ 中等                 │
│ 代码复杂度     │ 简单             │ 复杂（锁、竞态条件）   │
│ GIL 影响       │ 无                │ 部分缓解              │
│ 适用场景       │ 简单脚本/批处理   │ CPU+IO 混合           │
└───────────────┴──────────────────┴──────────────────────┘

┌──────────────────────────┐
│     异步（asyncio）       │
├──────────────────────────┤
│ 极低（单线程事件循环）     │
│ 极高（万~十万级）         │
│ 高（IO 时切换任务）       │
│ 中（需理解协程概念）       │
│ 完全绕过                  │
│ 高 IO 密集型服务/API      │
└──────────────────────────┘

时间线对比图

=== 同步（逐个执行，每个耗时 100ms） ===

请求 1: [==========][==========][========]
请求 2:                   [==========][==========][========]
请求 3:                                 [==========][==========][========]
总耗时: 300ms ───────────────────────────────────────────────────→

=== 多线程（并行执行） ===

Thread1: [==========][==========][========]
Thread2:               [==========][==========][========]
Thread3:                         [==========][==========][========]
总耗时: ~100ms ──────────────→

=== 异步（协作式切换） ===

Event Loop: [Req1:net][Req2:net][Req3:net][Req1:cpu][Req2:cpu][Req3:cpu]
            ← 单线程 →← 单线程 →← 单线程 →← 单线程 →← 单线程 →← 单线程 →
总耗时: ~70ms（重叠 IO 等待）→

核心概念

进程、线程、协程的关系

进程（Process）
├── 线程 Thread-1
│   ├── 协程 Coroutine-1
│   ├── 协程 Coroutine-2
│   └── 协程 Coroutine-3
├── 线程 Thread-2
│   ├── 协程 Coroutine-4
│   └── 协程 Coroutine-5
└── 线程 Thread-N
层级 调度者 切换成本 共享数据 数量级
进程 OS 内核 高（毫秒级） 通过 IPC 十到百
线程 OS 内核 中（微秒级） 共享内存 几十到几百
协程 用户空间（事件循环） 低（纳秒级） 共享变量 上万

关键术语速查

术语 英文 说明
协程 Coroutine 可暂停和恢复的函数
事件循环 Event Loop 协调所有协程运行的中枢
Future Future 表示将来结果的占位对象
Task Task 包装在 Event Loop 中的协程
非阻塞 IO Non-blocking IO 不等待的操作方式
IO 多路复用 IO Multiplexing 一个线程监控多个连接

asyncio 基础

Hello World

import asyncio

# 定义协程函数（用 async def）
async def hello():
    print("Hello")
    await asyncio.sleep(1)  # 异步等待 1 秒（不阻塞其他协程）
    print("World")

# 运行协程
asyncio.run(hello())
# Hello
# （等待 1 秒）
# World

基本 API 对照表

同步写法 异步写法 说明
time.sleep(1) await asyncio.sleep(1) 异步休眠
os.listdir() asyncio.to_thread(os.listdir) 将阻塞操作放入线程池
socket.accept() loop.sock_accept(sock) 异步接受连接
file.read() asyncio.to_thread(file.read) 异步读取文件

并发执行多个协程

gather —— 等全部完成

import asyncio
import time

async def fetch_page(name, delay):
    """模拟网络请求"""
    print(f"开始请求: {name}")
    await asyncio.sleep(delay)
    print(f"完成请求: {name}")
    return f"{name}: 响应数据 (耗时{delay}s)"

async def main_gather():
    start = time.time()
    
    # asyncio.gather: 并发执行，等待全部结果
    results = await asyncio.gather(
        fetch_page("page1", 1),
        fetch_page("page2", 2),
        fetch_page("page3", 0.5),
    )
    
    for i, r in enumerate(results, 1):
        print(f"  [{i}] {r}")
    
    total = time.time() - start
    print(f"\n总耗时: {total:.2f}s（并发比 3.5s 快很多）")

asyncio.run(main_gather())

as_completed —— 先完成先处理

import asyncio
import time

async def fetch_with_time(name, delay):
    await asyncio.sleep(delay)
    return name, delay

async def main_as_completed():
    start = time.time()
    
    coros = {
        'api_A': fetch_with_time('API-A', 3),
        'api_B': fetch_with_time('API-B', 1),
        'api_C': fetch_with_time('API-C', 2),
        'api_D': fetch_with_time('API-D', 0.5),
    }
    
    print("=== as_completed: 先完成的结果优先返回 ===")
    for coro in asyncio.as_completed(coros.values()):
        result = await coro
        elapsed = time.time() - start
        actual_coro_name = next(k for k, v in coros.items() if v == coro)
        print(f"  [{elapsed:.2f}s] ✓ {actual_coro_name} 完成")
    
    print(f"\n全部完成总耗时: {time.time() - start:.2f}s")

asyncio.run(main_as_completed())

create_task —— 启动即运行

import asyncio
import time

async def background_task(name, duration):
    """后台任务：持续运行"""
    for i in range(duration):
        print(f"[{name}] 工作进度: {i + 1}/{duration}")
        await asyncio.sleep(1)
    print(f"[{name}] 完成!")
    return f"{name}_completed"

async def main_create_task():
    start = time.time()
    
    # create_task: 立即启动，返回 Task 对象
    task1 = asyncio.create_task(background_task("备份", 3))
    task2 = asyncio.create_task(background_task("日志清理", 2))
    task3 = asyncio.create_task(background_task("缓存刷新", 4))
    
    # 主任务也可以做其他事
    print("[主程序] 执行其他工作...")
    await asyncio.sleep(1.5)
    print("[主程序] 检查后台任务状态:")
    print(f"  task1 done? {task1.done()}")
    print(f"  task2 done? {task2.done()}")
    
    # 等待全部完成
    results = await asyncio.gather(task1, task2, task3)
    print(f"\n所有后台任务完成: {results}")
    print(f"总耗时: {time.time() - start:.2f}s")

asyncio.run(main_create_task())

run_in_executor —— 调用阻塞代码

import asyncio
import os
import time

def blocking_cpu_work(n):
    """CPU 密集计算（不能用 await）"""
    total = sum(i * i for i in range(n))
    return total

def blocking_io_work(filepath):
    """文件 IO（不能用 await）"""
    time.sleep(2)  # 模拟慢 IO
    return f"File {filepath} processed"

async def main_run_in_executor():
    loop = asyncio.get_event_loop()
    
    # 将阻塞调用交给默认线程池
    result = await loop.run_in_executor(None, blocking_cpu_work, 1000000)
    print(f"CPU 计算结果: {result}")
    
    file_result = await loop.run_in_executor(None, blocking_io_work, "/tmp/test.txt")
    print(f"IO 结果: {file_result}")

asyncio.run(main_run_in_executor())

asyncio 实战：高性能 HTTP 客户端

import asyncio
import aiohttp
import time

async def fetch_url(session, url):
    """单个 URL 请求"""
    try:
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
            text = await resp.text()
            return url, len(text), resp.status
    except asyncio.TimeoutError:
        return url, 0, 'timeout'
    except Exception as e:
        return url, 0, str(e)

async def benchmark_concurrent(urls, limit=10):
    """并发基准测试"""
    start = time.time()
    semaphore = asyncio.Semaphore(limit)
    
    async with aiohttp.ClientSession() as session:
        tasks = []
        for url in urls:
            async with semaphore:
                tasks.append(fetch_url(session, url))
        results = await asyncio.gather(*tasks, return_exceptions=True)
    
    elapsed = time.time() - start
    print(f"URL 总数: {len(urls)} | 耗时: {elapsed:.2f}s | 吞吐: {len(urls)/elapsed:.1f} req/s")
    return results

if __name__ == '__main__':
    test_urls = [
        'https://httpbin.org/get',
        'https://httpbin.org/delay/1',
        'https://httpbin.org/delay/2',
        'https://httpbin.org/json',
        'https://httpbin.org/ip',
    ]
    asyncio.run(benchmark_concurrent(test_urls, limit=5))

asyncio 实战：TCP 服务器

import asyncio

class AsyncEchoServer:
    """异步 Echo 服务器"""
    
    def __init__(self, host='127.0.0.1', port=8888):
        self.host = host
        self.port = port
        self.clients = set()
    
    async def handle_client(self, reader, writer):
        addr = writer.get_extra_info('peername')
        print(f"✓ 新连接: {addr}")
        self.clients.add(writer)
        
        try:
            while True:
                data = await reader.readline()
                if not data:
                    break
                message = data.decode().strip()
                reply = f"ECHO: {message}".encode() + b'\n'
                
                for client in self.clients.copy():
                    if client != writer:
                        client.write(reply)
                        await client.drain()
                
                writer.write(reply)
                await writer.drain()
        except asyncio.CancelledError:
            print(f"✗ 连接断开: {addr}")
        finally:
            self.clients.discard(writer)
            writer.close()
            await writer.wait_closed()
    
    async def start(self):
        server = await asyncio.start_server(
            self.handle_client, self.host, self.port
        )
        print(f"✓ Echo 服务器运行于 {self.host}:{self.port}")
        async with server:
            await server.serve_forever()

# asyncio.run(AsyncEchoServer().start())

asyncio 实战：异步 Web 应用

import asyncio
import json

class AsyncApp:
    """极简异步 Web 应用"""
    
    def __init__(self):
        self.routes = {}
    
    def route(self, path, methods=None):
        if methods is None:
            methods = ['GET']
        def decorator(func):
            for method in methods:
                self.routes[(method.upper(), path)] = func
            return func
        return decorator
    
    async def _handle_request(self, method, path):
        handler = self.routes.get((method, path))
        if not handler:
            for (m, p), h in self.routes.items():
                if m != method:
                    continue
                pp = p.strip('/').split('/')
                cp = path.strip('/').split('/')
                if len(pp) != len(cp):
                    continue
                params = {}
                match = True
                for pi, ci in zip(pp, cp):
                    if pi.startswith('<') and pi.endswith('>'):
                        params[pi[1:-1]] = ci
                    elif pi != ci:
                        match = False
                        break
                if match:
                    handler = h
                    handler.__dict__['_params'] = params
        
        if handler:
            return await handler(method=method, path=path)
        else:
            return {'status': 404, 'body': json.dumps({'error': 'Not Found'})}
    
    async def handle_connection(self, reader, writer):
        try:
            request_line = await asyncio.wait_for(reader.readline(), timeout=5)
            if not request_line:
                writer.close()
                return
            
            parts = request_line.decode().strip().split(' ')
            method, path = parts[0], parts[1].split('?')[0]
            
            response = await self._handle_request(method, path)
            body = response.get('body', '{}').encode('utf-8')
            status = response.get('status', 200)
            
            headers = f"HTTP/1.1 {status} OK\r\nContent-Type: application/json\r\nContent-Length: {len(body)}\r\n\r\n"
            writer.write(headers.encode())
            writer.write(body)
            await writer.drain()
        except Exception as e:
            print(f"错误: {e}")
        finally:
            writer.close()
            await writer.wait_closed()

app = AsyncApp()

@app.route('/', methods=['GET'])
async def index(**kwargs):
    return {'status': 200, 'body': '<h1>⚡ 异步博客</h1><p>基于 asyncio 的高性能 Web</p>', 'content_type': 'text/html'}

@app.route('/api/posts', methods=['GET'])
async def api_posts(**kwargs):
    posts = [{'id': 1, 'title': 'Python 异步编程指南'}, {'id': 2, 'title': '协程与线程的区别'}]
    return {'status': 200, 'body': json.dumps(posts, ensure_ascii=False)}

@app.route('/post/<id>', methods=['GET'])
async def post_detail(**kwargs):
    pid = kwargs.get('_params', {}).get('id', '?')
    return {'status': 200, 'body': json.dumps({'id': int(pid), 'title': f'文章 {pid}'})}

if __name__ == '__main__':
    async def main():
        server = await asyncio.start_server(app.handle_connection, '127.0.0.1', 8899)
        print("✓ 异步博客服务器运行于 http://127.0.0.1:8899")
        async with server:
            await server.serve_forever()
    asyncio.run(main())

aiohttp 异步 Web 框架

"""
安装: pip install aiohttp
这是 asyncio 生态中最成熟的异步 Web 框架
"""

from aiohttp import web
import asyncio

async def homepage(request):
    return web.Response(text='<h1>你好，Aiohttp!</h1>')

async def get_user(request):
    user_id = request.match_info.get('id', '?')
    return web.json_response({'id': user_id, 'name': '张三'})

async def list_users(request):
    users = [{'id': i, 'name': f'用户{i}'} for i in range(1, 4)]
    return web.json_response(users)

@web.middleware
async def timing_middleware(request, handler):
    start = asyncio.get_event_loop().time()
    response = await handler(request)
    duration = asyncio.get_event_loop().time() - start
    print(f"[{request.method}] {request.path} → {response.status} ({duration:.3f}s)")
    return response

async def main():
    app = web.Application(middlewares=[timing_middleware])
    app.router.add_get('/', homepage)
    app.router.add_get('/users', list_users)
    app.router.add_get('/users/{id}', get_user)
    
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, 'localhost', 8099)
    await site.start()
    print("✓ Aiohttp 运行于 http://localhost:8099")
    
    try:
        while True:
            await asyncio.sleep(3600)
    except KeyboardInterrupt:
        await runner.cleanup()

# asyncio.run(main())

异常处理与超时控制

全面异常处理

import asyncio

async def risky_operation(task_name, should_fail=False):
    await asyncio.sleep(0.5)
    if should_fail:
        raise ValueError(f"{task_name} 出错！")
    return f"{task_name} 成功"

async def main_exception_handling():
    # 方式一：try-except（推荐）
    print("=== try-except ===")
    try:
        result = await risky_operation("任务A")
        print(f"  成功: {result}")
    except ValueError as e:
        print(f"  捕获异常: {e}")
    except asyncio.TimeoutError:
        print(f"  超时")
    except asyncio.CancelledError:
        print(f"  被取消")
    except Exception as e:
        print(f"  未知错误: {e}")
    
    # 方式二：ensure_future + add_done_callback
    print("\n=== done callback ===")
    future = asyncio.ensure_future(risky_operation("回调任务", should_fail=True))
    future.add_done_callback(lambda fut: print(f"  回调结果: {fut.result()}" if not fut.exception() else f"  回调异常: {fut.exception()}"))
    await asyncio.sleep(1)
    
    # 方式三：asyncio.wait + return_exceptions
    print("\n=== wait with return_exceptions ===")
    tasks = [risky_operation("OK任务"), risky_operation("Fail任务", should_fail=True)]
    done, pending = await asyncio.wait(tasks, return_exceptions=True)
    for d in done:
        if isinstance(d, Exception):
            print(f"  异常: {d}")
        else:
            print(f"  结果: {d}")
    for p in pending:
        p.cancel()

asyncio.run(main_exception_handling())

超时控制

import asyncio

async def slow_operation(name, delay):
    await asyncio.sleep(delay)
    return f"{name} 完成"

async def main_timeout():
    # 方式一：asyncio.wait_for
    print("=== wait_for ===")
    try:
        result = await asyncio.wait_for(slow_operation("快速操作", 0.5), timeout=10)
        print(f"  成功: {result}")
    except asyncio.TimeoutError:
        print(f"  超时！")
    
    try:
        result = await asyncio.wait_for(slow_operation("慢操作", 5), timeout=1)
    except asyncio.TimeoutError:
        print(f"  慢操作超时！(只给了 1s)")
    
    # 方式二：asyncio.shield（保护关键操作）
    print("\n=== shield ===")
    async def critical_db_write():
        await asyncio.sleep(2)
        print("  ✓ 数据库写入完成（不能被取消）")
        return True
    
    task = asyncio.create_task(critical_db_write())
    shielded = asyncio.shield(task)
    try:
        await asyncio.wait_for(shielded, timeout=1)
    except asyncio.TimeoutError:
        print("  主协程超时，但 shield 任务仍在执行")
        await task
        print("  ✓ 最终等待完成")

asyncio.run(main_timeout())

信号量控制并发

import asyncio
import time

async def database_query(query_id, delay):
    await asyncio.sleep(delay)
    return f"结果 {query_id}"

async def main_semaphore():
    start = time.time()
    sem = asyncio.Semaphore(3)  # 最多 3 个并发
    
    async def bounded_query(qid, delay):
        async with sem:
            result = await database_query(qid, delay)
            return result
    
    queries = [bounded_query(i, 0.5) for i in range(1, 11)]
    results = await asyncio.gather(*queries)
    print(f"{len(results)} 个查询完成，总耗时: {time.time()-start:.2f}s")

asyncio.run(main_semaphore())

生产者-消费者模式

import asyncio
import random

async def producer(queue, n):
    for i in range(1, n + 1):
        item = {'id': i, 'data': f'item_{i}', 'weight': random.uniform(0.1, 1.0)}
        await queue.put(item)
        print(f"→ 生产: {item['data']}")
        await asyncio.sleep(0.1)
    await queue.put(None)  # 终止信号
    print(f"✓ 生产者结束，共 {n} 项")

async def consumer(queue, worker_id):
    while True:
        item = await queue.get()
        if item is None:
            print(f"  [W{worker_id}] 收到终止信号，退出")
            break
        print(f"  [W{worker_id}] 处理: {item['data']}")
        await asyncio.sleep(item['weight'] * 0.5)
        queue.task_done()

async def main_producer_consumer():
    queue = asyncio.Queue(maxsize=5)
    num_items = 15
    consumers = [asyncio.create_task(consumer(queue, i + 1)) for i in range(3)]
    await producer(queue, num_items)
    await asyncio.gather(*consumers)
    print("✓ 所有数据处理完毕")

asyncio.run(main_producer_consumer())

asyncio 最佳实践

常见陷阱与解决方案

import asyncio

# ========== 陷阱 1: 忘记 await ==========
async def good_example():
    await asyncio.sleep(1)  # ✓ 正确
    print("等待完成")

async def bad_example():
    asyncio.sleep(1)  # ✗ 忘记 await，协程被丢弃！
    print("这会在睡眠之前立即执行！")

# ========== 陷阱 2: 在 sync 中调用 async ==========
# def sync_function():
#     asyncio.sleep(1)  # ❌ RuntimeError

# ========== 陷阱 3: 阻塞事件循环 ==========
async def blocking_hack():
    # time.sleep(1)  # ❌ 会阻塞整个事件循环！
    await asyncio.sleep(1)  # ✓ 正确做法

# ========== 陷阱 4: 全局 get_event_loop ==========
async def modern_approach():
    # old_way: loop = asyncio.get_event_loop(); loop.run_until_complete()
    # new way: asyncio.run(my_coroutine())
    pass

# ========== 陷阱 5: cancel 不当导致资源泄漏 ==========
async def careful_cancellation():
    lock = asyncio.Lock()
    async with lock:
        try:
            await asyncio.sleep(10)
        except asyncio.CancelledError:
            print("被取消了，但锁已安全释放")
            raise  # 重新抛出 CancelledError

# ========== 性能优化建议 ==========
async def performance_tips():
    # 1. 批量 IO：用 gather 代替逐个 await
    # 2. 限流：用 Semaphore 避免打挂下游
    # 3. 避免阻塞调用：time.sleep() → asyncio.sleep()
    # 4. 复用 Connection Pool
    pass

asyncio.run(performance_tips())

async/await 性能基准测试

import asyncio
import aiohttp
import time

def sync_benchmark():
    """模拟同步请求（串行）"""
    start = time.time()
    for i in range(5):
        time.sleep(0.5)
    elapsed = time.time() - start
    print(f"同步（串行）: {elapsed:.2f}s, 吞吐: {5/elapsed:.1f} req/s")

async def async_fetch(session, url, idx):
    try:
        async with session.get(url) as resp:
            await resp.text()
            return idx
    except Exception:
        return idx

async def async_benchmark():
    """异步请求（并行）"""
    start = time.time()
    urls = ['https://httpbin.org/delay/0.5'] * 5
    async with aiohttp.ClientSession() as session:
        await asyncio.gather(*(async_fetch(session, u, i) for i, u in enumerate(urls)))
    elapsed = time.time() - start
    print(f"异步（并发）: {elapsed:.2f}s, 吞吐: {5/elapsed:.1f} req/s")

# sync_benchmark()
# asyncio.run(async_benchmark())

小结

知识点 说明
async def 定义协程函数
await 暂停当前协程，让出控制权
asyncio.run() Python 3.7+ 入口函数
asyncio.gather() 并发执行，收集全部结果
asyncio.as_completed() 按完成顺序返回
asyncio.create_task() 立即启动，返回 Task 对象
asyncio.run_in_executor() 在线程池中运行阻塞代码
asyncio.Semaphore 控制并发数量
asyncio.Queue 生产者-消费者通信管道
asyncio.wait_for() 设置超时
asyncio.shield() 保护关键操作不被取消
aiohttp 异步 HTTP 客户端/服务端
aiohttp.web 异步 Web 框架

核心理解：异步编程的本质是在 IO 等待时切换任务。当协程执行 await 时，事件循环可以切换到其他就绪的协程执行，等到 IO 完成后再接回来。这使得单线程可以同时管理数万网络连接，非常适合高并发 IO 密集型场景。

练习题

  1. 实现一个分布式爬虫：给定 50 个 URL 列表，使用 aiohttp + asyncio.Semaphore 并发抓取，统计各页面的状态码、大小和标题。要求限制最大并发数为 10。

  2. 编写一个异步聊天室：支持多人连接，使用 asyncio.start_server + Queue 实现消息广播，并添加 /quit 命令优雅退出、/list 命令查看在线人数。

  3. 实现一个重试中间件：封装协程，自动对失败的操作进行指数退避重试（最多 3 次，间隔 1s、2s、4s），并通过 @retry(max_retries=3) 装饰器方式使用。