异步编程与 asyncio

异步编程概述

asyncio 是 Python 用于编写并发代码的标准库，使用 async/await 语法。它是处理 I/O 密集型高并发任务的最佳方案——比如 Web 服务器、爬虫、API 调用等。

与多线程不同，asyncio 在单线程内实现并发，没有线程切换开销和竞态条件问题。与多进程不同，asyncio 可以轻松处理数万个并发连接。

核心概念

协程（Coroutine）

协程是可以暂停执行和恢复执行的函数，是 async/await 的基础：

import asyncio

# 定义协程函数
async def say_hello():
    print("你好")
    await asyncio.sleep(1)  # 模拟 I/O 等待
    print("世界")

# 调用协程函数返回协程对象，不会执行
coro = say_hello()
print(type(coro))  # <class 'coroutine'>

# 运行协程
asyncio.run(say_hello())

关键区别：

普通函数调用：func() → 立即执行并返回结果
协程函数调用：async_func() → 返回协程对象，需要 await 或 asyncio.run

await 关键字

await 用于等待一个 awaitable 对象（协程、Task、Future）完成：

import asyncio

async def fetch_data():
    print("开始获取数据...")
    await asyncio.sleep(2)  # 模拟网络请求
    return {"data": "some data"}

async def main():
    # await 会暂停当前协程，等待 fetch_data 完成
    result = await fetch_data()
    print(f"结果: {result}")

asyncio.run(main())

事件循环

事件循环是 asyncio 的核心调度器，它负责管理所有协程的执行：

import asyncio

# Python 3.10+ 推荐用法
async def main():
    print("协程1 开始")
    await asyncio.sleep(1)
    print("协程1 结束")

async def main2():
    print("协程2 开始")
    await asyncio.sleep(0.5)
    print("协程2 结束")

# 获取事件循环
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

try:
    loop.run_until_complete(main())
finally:
    loop.close()

# 更简单的做法（Python 3.7+）
asyncio.run(main())

事件循环工作原理：

将协程注册到事件循环
事件循环按顺序执行协程
遇到 await 时暂停当前协程，切换到其他协程
I/O 操作完成后，恢复等待的协程

并发运行多个协程

asyncio.gather()

gather 并发执行多个协程并等待所有完成：

import asyncio
import time

async def fetch_user(user_id: int) -> dict:
    """模拟获取用户信息"""
    await asyncio.sleep(1)  # 模拟网络延迟
    return {"id": user_id, "name": f"用户{user_id}"}

async def fetch_orders(user_id: int) -> list:
    """模拟获取用户订单"""
    await asyncio.sleep(1.5)
    return [{"order_id": 100 + user_id, "amount": 99.9}]

async def main():
    start = time.perf_counter()

    # 并发执行
    user_task = fetch_user(1)
    orders_task = fetch_orders(1)

    # gather 等待所有协程完成
    user, orders = await asyncio.gather(user_task, orders_task)

    print(f"用户: {user}")
    print(f"订单: {orders}")
    print(f"耗时: {time.perf_counter() - start:.2f}秒")
    # 耗时约 1.5 秒（而非 2.5 秒）

asyncio.run(main())

asyncio.create_task()

create_task 将协程包装为 Task 并在后台调度：

import asyncio
import time

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

async def main():
    start = time.perf_counter()

    # 创建任务（立即开始后台执行）
    task1 = asyncio.create_task(slow_operation("任务1", 2))
    task2 = asyncio.create_task(slow_operation("任务2", 1))

    # 此时两个任务已在后台运行

    # 等待任务完成
    result1 = await task1
    result2 = await task2

    print(result1)  # 任务1 完成
    print(result2)  # 任务2 完成
    print(f"耗时: {time.perf_counter() - start:.2f}秒")

asyncio.run(main())

asyncio.create_task vs asyncio.gather

import asyncio

# create_task：适合需要单独管理的独立任务
async def main1():
    task = asyncio.create_task(some_coro())
    # 处理其他事情...
    result = await task
    return result

# gather：适合等待一组相关协程全部完成
async def main2():
    results = await asyncio.gather(
        fetch(1),
        fetch(2),
        fetch(3),
    )
    return results

awaitable 对象的三种类型

1. 协程（Coroutine）

async def my_coro():
    return 42

coro = my_coro()  # 协程对象
result = await coro  # 可以 await

2. Task

async def my_coro():
    return 42

task = asyncio.create_task(my_coro())  # Task 对象
result = await task  # 可以 await

3. Future

# Future 是更低层的 awaitable
# Task 是 Future 的子类
# 通常不需要直接使用 Future
future = asyncio.Future()
# 在某个时候 future.set_result(value)
result = await future

asyncio.sleep

asyncio.sleep 是异步版本的 time.sleep，它不会阻塞线程：

import asyncio
import time

async def demo_sleep():
    # BAD: 会阻塞整个线程
    # time.sleep(1)

    # GOOD: 只暂停当前协程，其他协程可以运行
    await asyncio.sleep(1)

async def task(name: str, delay: float):
    print(f"{name} 开始")
    await asyncio.sleep(delay)  # 非阻塞等待
    print(f"{name} 结束 (等待了{delay}秒)")

async def main():
    # 同时启动3个任务
    await asyncio.gather(
        task("A", 2),
        task("B", 1),
        task("C", 3),
    )

start = time.perf_counter()
asyncio.run(main())
print(f"总耗时: {time.perf_counter() - start:.2f}秒")
# 输出:
# A 开始
# B 开始
# C 开始
# B 结束 (等待了1秒)
# A 结束 (等待了2秒)
# C 结束 (等待了3秒)
# 总耗时: 3.00秒（而非6秒）

异步上下文管理器

import asyncio

class AsyncResource:
    """异步资源管理器"""
    async def __aenter__(self):
        print("获取资源...")
        await asyncio.sleep(0.5)
        print("资源就绪")
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("释放资源...")
        await asyncio.sleep(0.3)
        print("资源已释放")

    async def use(self):
        print("使用资源中...")
        await asyncio.sleep(0.5)

async def main():
    async with AsyncResource() as resource:
        await resource.use()

asyncio.run(main())

使用 @asynccontextmanager

from contextlib import asynccontextmanager

@asynccontextmanager
async def database_session():
    """异步数据库会话上下文管理器"""
    print("连接数据库...")
    await asyncio.sleep(0.5)
    session = {"connected": True}
    try:
        yield session  # 提供会话
    finally:
        print("关闭数据库连接...")
        await asyncio.sleep(0.3)

async def main():
    async with database_session() as session:
        print(f"使用数据库: {session}")
        await asyncio.sleep(0.5)

asyncio.run(main())

异步迭代器

import asyncio

class AsyncRange:
    """异步版本的 range"""
    def __init__(self, start, end, delay=0.5):
        self.start = start
        self.end = end
        self.delay = delay

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.start >= self.end:
            raise StopAsyncIteration
        value = self.start
        self.start += 1
        await asyncio.sleep(self.delay)  # 模拟异步操作
        return value

async def main():
    async for num in AsyncRange(1, 5, 0.3):
        print(f"收到: {num}")

asyncio.run(main())

异步生成器

import asyncio

async def async_range(start: int, end: int, delay: float = 0.5):
    """异步生成器"""
    for i in range(start, end):
        await asyncio.sleep(delay)
        yield i

async def main():
    async for num in async_range(1, 6, 0.3):
        print(f"生成: {num}")

asyncio.run(main())

asyncio.Queue

import asyncio
import random

async def producer(queue: asyncio.Queue, item_count: int):
    """生产者"""
    for i in range(item_count):
        item = f"数据-{i}"
        await queue.put(item)
        print(f"生产: {item}")
        await asyncio.sleep(random.uniform(0.1, 0.3))
    # 发送结束信号
    await queue.put(None)

async def consumer(queue: asyncio.Queue, name: str):
    """消费者"""
    while True:
        item = await queue.get()
        if item is None:
            await queue.put(None)  # 传递给其他消费者
            break
        print(f"  {name} 消费: {item}")
        await asyncio.sleep(random.uniform(0.2, 0.5))

async def main():
    queue = asyncio.Queue(maxsize=5)

    # 创建生产者消费者任务
    tasks = [
        asyncio.create_task(producer(queue, 8)),
        asyncio.create_task(consumer(queue, "消费者A")),
        asyncio.create_task(consumer(queue, "消费者B")),
    ]

    await asyncio.gather(*tasks)

asyncio.run(main())

运行阻塞代码

在异步代码中，如果遇到阻塞操作，使用 run_in_executor 将其交给线程池：

import asyncio
import time

def blocking_io() -> str:
    """同步阻塞函数"""
    time.sleep(2)  # 假设是阻塞 I/O
    return "阻塞操作结果"

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

    # 将阻塞函数放到线程池执行
    result = await loop.run_in_executor(
        None,  # None = 默认线程池
        blocking_io
    )
    print(result)

asyncio.run(main())

封装同步库为异步接口

import asyncio
import requests  # 同步库

async def async_request(url: str) -> dict:
    """将同步 requests 封装为异步"""
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(
        None,
        requests.get,
        url
    )

# 更好的方式：直接使用 aiohttp

asyncio.wait 和 as_completed

asyncio.wait

import asyncio

async def make_request(url: str) -> str:
    await asyncio.sleep(1)
    return f"响应来自 {url}"

async def main():
    tasks = [
        asyncio.create_task(make_request(f"http://api.example.com/{i}"))
        for i in range(5)
    ]

    # 等待所有完成
    done, pending = await asyncio.wait(tasks)

    for task in done:
        print(task.result())

    # 返回结果的方式
    # FIRST_COMPLETED: 任一完成就返回
    # FIRST_EXCEPTION: 任一异常就返回
    # ALL_COMPLETED: 全部完成（默认）
    done, pending = await asyncio.wait(
        tasks,
        timeout=5.0,
        return_when=asyncio.FIRST_COMPLETED
    )

asyncio.run(main())

asyncio.as_completed

import asyncio

async def work(n: int) -> str:
    await asyncio.sleep(n)
    return f"任务 {n} 完成（耗时{n}秒）"

async def main():
    tasks = [work(3), work(1), work(2)]

    # 按完成顺序处理结果
    for coro in asyncio.as_completed(tasks):
        result = await coro
        print(f"最先完成: {result}")

asyncio.run(main())
# 输出:
# 最先完成: 任务 1 完成（耗时1秒）
# 最先完成: 任务 2 完成（耗时2秒）
# 最先完成: 任务 3 完成（耗时3秒）

超时与取消

asyncio.wait_for

import asyncio

async def slow_operation():
    await asyncio.sleep(10)
    return "完成"

async def main():
    try:
        # 设置超时，超过时间引发 TimeoutError
        result = await asyncio.wait_for(slow_operation(), timeout=2)
        print(result)
    except asyncio.TimeoutError:
        print("操作超时！")

asyncio.run(main())

任务取消

import asyncio

async def cancellable_work():
    try:
        for i in range(10):
            await asyncio.sleep(0.5)
            print(f"工作进度: {i}")
    except asyncio.CancelledError:
        print("任务被取消")
        raise  # 必须重新抛出

async def main():
    task = asyncio.create_task(cancellable_work())
    await asyncio.sleep(1.5)
    task.cancel()  # 取消任务

    try:
        await task
    except asyncio.CancelledError:
        print("已确认取消")

asyncio.run(main())

实战：异步 Web 爬虫

import asyncio
import aiohttp
from typing import Optional

# 需要安装: pip install aiohttp

async def fetch_url(
    session: aiohttp.ClientSession,
    url: str,
    timeout: int = 10
) -> Optional[dict]:
    """异步获取 URL 内容"""
    try:
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=timeout)) as response:
            text = await response.text()
            return {
                "url": url,
                "status": response.status,
                "length": len(text),
                "content": text[:200]  # 只保留前200字符
            }
    except Exception as e:
        return {"url": url, "error": str(e)}

async def crawl_urls(urls: list[str], max_concurrent: int = 5) -> list[dict]:
    """并发爬取多个 URL"""
    connector = aiohttp.TCPConnector(limit=max_concurrent)

    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        return [
            r for r in results
            if isinstance(r, dict)
        ]

async def main():
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2",
        "https://httpbin.org/delay/3",
        "https://httpbin.org/status/200",
        "https://httpbin.org/status/404",
    ]

    import time
    start = time.perf_counter()

    results = await crawl_urls(urls)

    elapsed = time.perf_counter() - start
    print(f"爬取完成，共 {len(results)} 个结果，耗时 {elapsed:.2f}秒")

    for result in results:
        if "error" in result:
            print(f"✗ {result['url']}: {result['error']}")
        else:
            print(f"✓ {result['url']}: status={result['status']}, size={result['length']}")

# asyncio.run(main())

实战：异步文件读取

import asyncio
import aiofiles

# 需要安装: pip install aiofiles

async def read_file_async(filepath: str) -> str:
    """异步读取文件"""
    async with aiofiles.open(filepath, "r", encoding="utf-8") as f:
        return await f.read()

async def write_file_async(filepath: str, content: str):
    """异步写入文件"""
    async with aiofiles.open(filepath, "w", encoding="utf-8") as f:
        await f.write(content)

async def process_files(filepaths: list[str]):
    """并发处理多个文件"""
    tasks = [read_file_async(fp) for fp in filepaths]
    contents = await asyncio.gather(*tasks)

    # 处理内容（例如：转为大写）
    processed = [content.upper() for content in contents]

    # 并发写入
    write_tasks = [
        write_file_async(f"processed_{i}.txt", content)
        for i, content in enumerate(processed)
    ]
    await asyncio.gather(*write_tasks)

# async def main():
#     await process_files(["file1.txt", "file2.txt", "file3.txt"])
# asyncio.run(main())

实战：同步 vs 异步性能对比

import asyncio
import time

# 同步版本
def sync_work(n: int) -> int:
    """模拟 I/O 操作"""
    time.sleep(0.5)  # 同步阻塞
    return n * 2

def sync_main():
    start = time.perf_counter()
    results = []
    for i in range(10):
        result = sync_work(i)
        results.append(result)
    elapsed = time.perf_counter() - start
    return results, elapsed

# 异步版本
async def async_work(n: int) -> int:
    """异步 I/O 操作"""
    await asyncio.sleep(0.5)  # 非阻塞
    return n * 2

async def async_main():
    start = time.perf_counter()
    tasks = [asyncio.create_task(async_work(i)) for i in range(10)]
    results = await asyncio.gather(*tasks)
    elapsed = time.perf_counter() - start
    return results, elapsed

def compare():
    # 同步
    sync_results, sync_time = sync_main()
    print(f"同步: {sync_time:.2f}秒")

    # 异步
    async_results, async_time = asyncio.run(async_main())
    print(f"异步: {async_time:.2f}秒")

    print(f"加速比: {sync_time / async_time:.0f}x")
    print(f"结果一致: {sync_results == async_results}")

compare()
# 输出:
# 同步: 5.00秒（10 × 0.5）
# 异步: 0.50秒（并发执行）
# 加速比: 10x
# 结果一致: True

常见陷阱

陷阱1：在异步函数中调用阻塞函数

import asyncio
import time

# BAD: 阻塞整个事件循环
async def bad_func():
    time.sleep(5)  # 阻塞了所有协程！

# GOOD: 使用 asyncio.sleep
async def good_func():
    await asyncio.sleep(5)

# GOOD: 或使用 run_in_executor
async def good_func2():
    await asyncio.get_running_loop().run_in_executor(None, time.sleep, 5)

陷阱2：忘记 await

import asyncio

async def fetch():
    await asyncio.sleep(1)
    return 42

async def main():
    # BAD: 没有 await，协程不会执行
    result = fetch()
    print(result)  # <coroutine object ...>

    # GOOD: 正确 await
    result = await fetch()
    print(result)  # 42

asyncio.run(main())

陷阱3：在事件循环外调用 async 函数

import asyncio

async def hello():
    print("你好")

# BAD: 必须通过事件循环执行
# hello()  # RuntimeWarning

# GOOD: 使用 asyncio.run
asyncio.run(hello())  # 正确

陷阱4：共享可变状态

import asyncio

# 虽然是单线程，但仍需注意：
# 在 await 之间，状态可能被其他协程修改

counter = 0

async def increment():
    global counter
    temp = counter
    await asyncio.sleep(0)  # 切换到其他协程！
    counter = temp + 1

async def main():
    await asyncio.gather(increment(), increment(), increment())
    print(counter)  # 可能是 1 或 2，不是 3！

# 解决方案：使用 asyncio.Lock
lock = asyncio.Lock()

async def safe_increment():
    global counter
    async with lock:
        temp = counter
        await asyncio.sleep(0)
        counter = temp + 1

async def main_safe():
    await asyncio.gather(safe_increment(), safe_increment(), safe_increment())
    print(counter)  # 3

asyncio.run(main_safe())

三种并发模型的对比

# 这是一个总结对比，不是完整可运行代码

# 1. 多线程（threading）
# 适用：I/O 密集型
# 优点：直观，适合有大量阻塞调用的场景
# 缺点：GIL 限制，有竞态条件
# 示例：网络请求、文件读写

# 2. 多进程（multiprocessing）
# 适用：CPU 密集型
# 优点：真正并行，不受 GIL 限制
# 缺点：创建开销大，通信复杂
# 示例：数值计算、图像处理

# 3. asyncio
# 适用：高并发 I/O
# 优点：单线程无锁，可处理数万连接
# 缺点：需要全栈异步，生态不完善
# 示例：Web 服务器、爬虫、API 网关

性能对比图（概念）

                    I/O 密集型         CPU 密集型
串行                  ██ 慢              ██ 慢
多线程              ██████████ 快       ██ 慢（GIL）
多进程              ██████ 中等        ██████████ 快
asyncio             ████████████ 最快   ██ 慢

Python 3.13+ 异步新特性

Python 3.13 中的 asyncio 改进：

import asyncio

# Python 3.12+: TaskGroup 简化任务管理
async def task_group_demo():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(asyncio.sleep(1))
        task2 = tg.create_task(asyncio.sleep(2))
        task3 = tg.create_task(asyncio.sleep(3))
    # 所有任务完成或出错时自动管理

# 错误传播：TaskGroup 中任一任务失败会取消其他任务
async def task_group_error():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(asyncio.sleep(1))
        tg.create_task(asyncio.sleep(2))
        tg.create_task(raise_error())  # 这个失败会取消所有任务

异步编程最佳实践

1. 使用 asyncio.run 作为入口

# Python 3.7+ 推荐
def main():
    asyncio.run(async_main())

asyncio.run(main())

2. 使用 TaskGroup 管理任务

# Python 3.11+
async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(task1())
        tg.create_task(task2())

3. 设置合理的超时

async def fetch_with_timeout(url: str):
    try:
        result = await asyncio.wait_for(fetch(url), timeout=5)
        return result
    except asyncio.TimeoutError:
        return None

4. 使用信号量控制并发量

semaphore = asyncio.Semaphore(10)  # 最多10个并发

async def limited_request(url: str):
    async with semaphore:
        return await fetch(url)

5. 避免在协程中混用同步 I/O

# BAD
async def bad():
    file = open("data.txt")  # 同步操作，阻塞事件循环
    ...

# GOOD
async def good():
    async with aiofiles.open("data.txt") as f:
        ...

小结

asyncio 是 Python 异步编程的核心，它通过单线程事件循环实现了高效的 I/O 并发。本章我们系统学习了协程与 async/await 语法、事件循环的工作原理、并发运行协程的三种方式（gather、create_task、as_completed）、awaitable 对象的三种类型、异步上下文管理器和异步迭代器、asyncio.Queue 的生产者-消费者模式、超时与任务取消机制。通过异步爬虫和性能对比等实战案例，你应该已经清楚 asyncio 的适用场景和优势。最后记住：asyncio 适合高并发 I/O 密集型任务，CPU 密集型用多进程，简单 I/O 用多线程——选择合适的工具，才能写出高效的并发程序。

Summary: async/await、事件循环、协程并发、TaskGroup 与异步实战。