异步编程详解

概述

Python 的异步编程基于 asyncio 库，通过 async/await 语法实现协程（coroutine）。异步编程特别适合 IO 密集型任务，如网络请求、数据库查询和文件操作，可以在等待 IO 时让出控制权执行其他任务，从而提高程序的并发性能。

基础概念

同步 vs 异步

同步代码按顺序执行，遇到 IO 操作时阻塞等待。异步代码在等待 IO 时可以切换到其他任务，充分利用等待时间。

# 同步：串行执行，总耗时 = 请求1 + 请求2
result1 = requests.get("/api/users")   # 等待 1 秒
result2 = requests.get("/api/posts")   # 等待 1 秒
# 总耗时约 2 秒

# 异步：并发执行，总耗时 = max(请求1, 请求2)
result1, result2 = await asyncio.gather(
    fetch("/api/users"),  # 等待 1 秒
    fetch("/api/posts"),  # 等待 1 秒
)
# 总耗时约 1 秒

协程与事件循环

协程是使用 async def 定义的函数，调用后返回协程对象，不会立即执行。事件循环负责调度协程的执行，在协程遇到 await 时切换到其他可运行的协程。

import asyncio

async def main():
    print("Hello")

# 获取/创建事件循环
loop = asyncio.get_event_loop()
loop.run_until_complete(main())

# 或使用简化 API（推荐）
asyncio.run(main())

快速上手

第一个异步程序

import asyncio

async def say_hello(name, delay):
    """异步函数：等待 delay 秒后打印问候"""
    await asyncio.sleep(delay)  # 非阻塞等待
    print(f"Hello, {name}!")

async def main():
    # 并发执行两个协程
    await asyncio.gather(
        say_hello("Alice", 2),
        say_hello("Bob", 1),
    )
    # 输出顺序：Bob（1秒后）, Alice（2秒后）

asyncio.run(main())

Task 与 Future

# Task：包装协程，使其可以被事件循环调度
task = asyncio.create_task(fetch_data())

# Future：低层级对象，表示未来才会有结果
future = loop.create_future()
future.set_result(42)

# 等待多个任务
results = await asyncio.gather(
    fetch_users(),
    fetch_posts(),
    return_exceptions=True  # 异常作为结果返回，不中断其他任务
)

详细用法

同步原语

# 锁：保护共享资源
lock = asyncio.Lock()
async with lock:
    await critical_section()

# 信号量：限制并发数
sem = asyncio.Semaphore(10)
async with sem:
    await limited_operation()

# 事件：通知机制
event = asyncio.Event()
await event.wait()  # 等待事件被设置
event.set()          # 设置事件，唤醒所有等待者

# 条件变量
cond = asyncio.Condition()
async with cond:
    await cond.wait()  # 等待通知
    # 或
    cond.notify_all()  # 通知所有等待者

超时控制

# 方式一：wait_for
try:
    result = await asyncio.wait_for(fetch(), timeout=5.0)
except asyncio.TimeoutError:
    print("操作超时")

# 方式二：timeout 上下文管理器（Python 3.11+）
async with asyncio.timeout(5.0):
    result = await fetch()

# 方式三：手动取消任务
task = asyncio.create_task(fetch())
await asyncio.sleep(5.0)
task.cancel()
try:
    await task
except asyncio.CancelledError:
    print("任务已取消")

异步迭代器与生成器

# 异步迭代器
class AsyncRange:
    def __init__(self, count):
        self.count = count

    def __aiter__(self):
        self.i = 0
        return self

    async def __anext__(self):
        if self.i >= self.count:
            raise StopAsyncIteration
        await asyncio.sleep(0.1)
        self.i += 1
        return self.i

# 使用 async for
async for item in AsyncRange(5):
    print(item)

# 异步生成器
async def async_count(n):
    for i in range(n):
        await asyncio.sleep(0.1)
        yield i

async for num in async_count(5):
    print(num)

异步队列

async def producer(queue, items):
    """生产者：向队列中放入数据"""
    for item in items:
        await queue.put(item)
        print(f"生产: {item}")
    # 放入结束标记
    await queue.put(None)

async def consumer(queue, name):
    """消费者：从队列中取出数据处理"""
    while True:
        item = await queue.get()
        if item is None:
            queue.task_done()
            break
        print(f"消费者 {name} 处理: {item}")
        await asyncio.sleep(0.1)  # 模拟处理
        queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=5)
    # 启动生产者和消费者
    await asyncio.gather(
        producer(queue, range(10)),
        consumer(queue, "A"),
        consumer(queue, "B"),
    )

asyncio.run(main())

常见场景

场景一：并发 HTTP 请求

import aiohttp

async def fetch_url(session, url):
    """异步获取 URL 内容"""
    async with session.get(url) as response:
        return await response.text()

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

# 使用
urls = ["https://api.example.com/users", "https://api.example.com/posts"]
results = await fetch_all(urls)

场景二：限制并发数

async def bounded_gather(coros, limit=10):
    """限制并发数的 gather"""
    sem = asyncio.Semaphore(limit)

    async def wrap(coro):
        async with sem:
            return await coro

    return await asyncio.gather(*[wrap(c) for c in coros])

# 使用
tasks = [fetch_data(i) for i in range(100)]
results = await bounded_gather(tasks, limit=20)  # 最多 20 个并发

场景三：优雅关闭

async def main():
    tasks = [asyncio.create_task(worker(i)) for i in range(5)]

    try:
        # 等待所有任务完成
        await asyncio.gather(*tasks)
    except asyncio.CancelledError:
        # 收到取消信号时，等待任务完成
        for task in tasks:
            task.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)

注意事项

• async 函数必须用 await 调用，否则只是创建了协程对象，不会执行
• 不要在异步代码中使用同步的阻塞调用（如 time.sleep、requests.get），会阻塞整个事件循环
• 使用 asyncio.to_thread() 在线程中执行阻塞操作
• asyncio 不是多线程，同一时刻只有一个协程在执行
• CPU 密集型任务应使用多进程（multiprocessing），而非 asyncio
• asyncio.run() 每次调用都会创建新的事件循环，不要嵌套调用

进阶用法

在异步代码中调用同步函数

import asyncio

def blocking_io():
    """阻塞的同步函数"""
    import time
    time.sleep(1)
    return "完成"

async def main():
    # 在线程池中执行阻塞函数
    result = await asyncio.to_thread(blocking_io)
    print(result)

asyncio.run(main())

异步上下文管理器

from contextlib import asynccontextmanager

@asynccontextmanager
async def async_db_connection(url):
    """异步数据库连接"""
    conn = await connect(url)
    try:
        yield conn
    finally:
        await conn.close()

async with async_db_connection("postgresql://...") as db:
    await db.execute("SELECT 1")

异步信号处理

import signal

async def main():
    loop = asyncio.get_running_loop()
    stop = loop.create_future()

    # 注册信号处理
    loop.add_signal_handler(signal.SIGINT, stop.set_result, None)
    loop.add_signal_handler(signal.SIGTERM, stop.set_result, None)

    print("服务启动，按 Ctrl+C 停止")
    await stop  # 等待信号
    print("正在关闭...")

asyncio.run(main())

asyncio 与多进程结合

import asyncio
from concurrent.futures import ProcessPoolExecutor

def cpu_heavy(n):
    """CPU 密集型计算"""
    return sum(i * i for i in range(n))

async def main():
    loop = asyncio.get_running_loop()

    # 在进程池中执行 CPU 密集型任务
    with ProcessPoolExecutor() as pool:
        results = await asyncio.gather(
            loop.run_in_executor(pool, cpu_heavy, 10**7),
            loop.run_in_executor(pool, cpu_heavy, 10**7),
        )
    print(results)

asyncio.run(main())