Python 异步编程深度实战：TaskGroup、取消传播、背压与线程边界全解析

异步不是银弹，但用对了，它能让你的服务在同样的硬件上多撑住十倍的并发。用错了，你会得到一堆难以调试的"幽灵 bug"。

---

一、为什么 asyncio 值得你认真对待？

2024 年的 Python 后端开发，asyncio 已经不是"加分项"，而是基础设施。FastAPI、aiohttp、Starlette、Celery 6.x、SQLAlchemy 2.0 的异步模式……几乎所有主流框架都在向协程靠拢。

但很多开发者对 asyncio 的理解停留在"会写 async/await"的层面。一旦遇到任务取消、并发控制、CPU 密集型混合场景，就开始出现各种诡异问题：任务莫名消失、服务内存持续增长、线程池把 event loop 卡死。

这篇文章，我们从 TaskGroup 出发，把异步编程里最硬核的几个概念逐一拆透。

---

二、协程、任务、Future：先把概念理清

很多混乱来自概念不清。在动手之前，先把三个核心对象的关系说明白：

协程函数 (async def)
    │ 调用后得到
    ▼
协程对象 (coroutine)
    │ 被 asyncio.create_task() 或 await 包装后
    ▼
Task (Future 的子类)
    │ 被 event loop 调度执行
    ▼
Future (底层承诺对象，代表"将来会有结果")

import asyncio

async def greet(name: str) -> str:
    await asyncio.sleep(0.1)
    return f"Hello, {name}"

async def main():
    # 方式1：直接 await，串行执行
    result = await greet("Alice")  # 等待完成才继续

    # 方式2：create_task，并发执行
    task = asyncio.create_task(greet("Bob"))  # 立即提交给 event loop
    # ... 可以做其他事 ...
    result = await task  # 需要结果时再等待

    # 关键区别：Task 一旦创建就开始运行，协程对象不 await 就什么都不做
    coro = greet("Charlie")  # 只是创建了协程对象，没有执行
    # 如果忘记 await，Python 3.12+ 会发出 RuntimeWarning

asyncio.run(main())

---

三、TaskGroup：结构化并发的正确姿势

Python 3.11 引入的 TaskGroup 是 asyncio 近年来最重要的改进，它解决了 gather 的一个根本性缺陷。

3.1 gather 的问题

import asyncio

async def risky_worker(name: str, should_fail: bool):
    await asyncio.sleep(0.5)
    if should_fail:
        raise ValueError(f"{name} 出错了")
    return f"{name} 完成"

async def with_gather():
    try:
        results = await asyncio.gather(
            risky_worker("A", False),
            risky_worker("B", True),   # 这个会失败
            risky_worker("C", False),
            return_exceptions=False    # 默认行为
        )
    except ValueError as e:
        print(f"捕获到异常: {e}")
        # 问题：A 和 C 的任务状态如何？
        # 答案：它们仍在后台运行，你已经失去了对它们的控制
        # 这就是"任务泄漏"

asyncio.run(with_gather())

3.2 TaskGroup 的结构化并发

import asyncio

async def worker(name: str, delay: float, should_fail: bool = False) -> str:
    print(f"{name} 开始")
    await asyncio.sleep(delay)
    if should_fail:
        raise ValueError(f"{name} 执行失败")
    print(f"{name} 完成")
    return f"{name} done"

async def with_task_group():
    try:
        async with asyncio.TaskGroup() as tg:
            t1 = tg.create_task(worker("A", 1.0))
            t2 = tg.create_task(worker("B", 0.5, should_fail=True))
            t3 = tg.create_task(worker("C", 2.0))
            # TaskGroup 的保证：
            # 1. 任意一个任务失败，其余任务会被自动取消
            # 2. 退出 with 块时，所有任务必然已完成（成功/失败/取消）
            # 3. 所有异常会被收集，以 ExceptionGroup 形式抛出
    except* ValueError as eg:
        # Python 3.11+ 的 except* 语法，专门处理 ExceptionGroup
        print(f"捕获到 {len(eg.exceptions)} 个 ValueError:")
        for exc in eg.exceptions:
            print(f"  - {exc}")

asyncio.run(with_task_group())
# 输出：
# A 开始
# B 开始
# C 开始
# 捕获到 1 个 ValueError:
#   - B 执行失败
# （A 和 C 被自动取消，没有任务泄漏）

3.3 嵌套 TaskGroup：构建任务树

async def fetch_user_data(user_id: int) -> dict:
    """模拟获取用户完整数据：并发获取各子项"""
    async with asyncio.TaskGroup() as tg:
        profile_task = tg.create_task(fetch_profile(user_id))
        orders_task  = tg.create_task(fetch_orders(user_id))
        settings_task = tg.create_task(fetch_settings(user_id))
    
    # 到这里，三个任务都已完成
    return {
        "profile":  profile_task.result(),
        "orders":   orders_task.result(),
        "settings": settings_task.result(),
    }

async def fetch_dashboard(user_ids: list[int]) -> list[dict]:
    """并发获取多个用户的数据"""
    async with asyncio.TaskGroup() as tg:
        tasks = [
            tg.create_task(fetch_user_data(uid))
            for uid in user_ids
        ]
    return [t.result() for t in tasks]

---

四、取消传播：理解 CancelledError 的传递机制

取消是 asyncio 里最容易踩坑的地方。很多开发者不理解取消是如何传播的，导致要么取消不干净，要么把正常的取消当成错误处理。

4.1 取消的基本机制

import asyncio

async def cancellable_worker():
    try:
        print("开始工作")
        await asyncio.sleep(10)  # 模拟长时间操作
        print("工作完成")
    except asyncio.CancelledError:
        print("收到取消信号，执行清理...")
        # 关键：清理完成后，必须重新抛出 CancelledError
        # 否则取消信号会被"吞掉"，导致取消传播中断
        raise  # 不能省略！
    finally:
        print("finally 块始终执行（无论正常还是取消）")

async def main():
    task = asyncio.create_task(cancellable_worker())
    await asyncio.sleep(0.1)  # 让 worker 开始运行
    
    task.cancel()  # 发送取消请求
    
    try:
        await task
    except asyncio.CancelledError:
        print(f"任务已取消，状态: {task.cancelled()}")

asyncio.run(main())
# 输出：
# 开始工作
# 收到取消信号，执行清理...
# finally 块始终执行（无论正常还是取消）
# 任务已取消，状态: True

4.2 取消屏蔽：asyncio.shield

async def critical_cleanup(resource_id: str):
    """关键清理操作，不允许被取消中断"""
    print(f"开始清理资源 {resource_id}")
    await asyncio.sleep(1)  # 模拟清理耗时
    print(f"资源 {resource_id} 清理完成")

async def worker_with_shield(resource_id: str):
    try:
        await asyncio.sleep(5)  # 主要工作
    except asyncio.CancelledError:
        # 即使 worker 被取消，也要确保清理完成
        await asyncio.shield(critical_cleanup(resource_id))
        raise  # 清理完成后再传播取消

async def main():
    task = asyncio.create_task(worker_with_shield("DB-连接池"))
    await asyncio.sleep(0.5)
    task.cancel()
    
    try:
        await task
    except asyncio.CancelledError:
        print("任务取消，但清理已完成")

asyncio.run(main())

4.3 带超时的取消：asyncio.timeout

import asyncio

async def slow_api_call(endpoint: str) -> str:
    await asyncio.sleep(5)  # 模拟慢接口
    return f"{endpoint} 响应"

async def fetch_with_timeout():
    # Python 3.11+ 推荐方式，比 wait_for 更清晰
    try:
        async with asyncio.timeout(2.0):  # 2秒超时
            result = await slow_api_call("/api/data")
            return result
    except TimeoutError:
        print("请求超时，使用降级数据")
        return {"status": "degraded", "data": []}

# 多个操作共享同一个超时预算
async def fetch_with_deadline():
    deadline = asyncio.get_event_loop().time() + 3.0  # 3秒总预算
    
    async with asyncio.timeout_at(deadline):
        # 这两个操作共享 3 秒预算
        part1 = await slow_api_call("/api/part1")  # 如果这个花了2.5秒
        part2 = await slow_api_call("/api/part2")  # 这个只剩0.5秒

---

五、背压：不让生产者把消费者淹没

背压（Backpressure）是高并发系统里的核心概念。生产者速度 > 消费者速度时，如果没有背压机制，队列会无限增长，最终 OOM。

5.1 用 asyncio.Queue 实现背压

import asyncio
import random

async def producer(queue: asyncio.Queue, name: str, count: int):
    """生产者：生成任务并放入队列"""
    for i in range(count):
        item = f"{name}-任务{i}"
        
        # put() 在队列满时会自动等待（背压的核心）
        await queue.put(item)
        print(f"[生产] {item}，队列大小: {queue.qsize()}")
        
        await asyncio.sleep(random.uniform(0.01, 0.05))  # 模拟生产速度
    
    print(f"{name} 生产完毕")

async def consumer(queue: asyncio.Queue, name: str):
    """消费者：从队列取任务并处理"""
    while True:
        try:
            # 等待任务，超时则退出
            item = await asyncio.wait_for(queue.get(), timeout=2.0)
            
            print(f"[消费] {name} 处理: {item}")
            await asyncio.sleep(random.uniform(0.05, 0.15))  # 消费比生产慢
            
            queue.task_done()  # 标记任务完成
        except TimeoutError:
            print(f"{name} 超时退出")
            break

async def pipeline_demo():
    # maxsize=10 是关键：队列满了，生产者会被阻塞
    queue = asyncio.Queue(maxsize=10)
    
    async with asyncio.TaskGroup() as tg:
        # 2个生产者，3个消费者
        tg.create_task(producer(queue, "生产者A", 20))
        tg.create_task(producer(queue, "生产者B", 20))
        
        for i in range(3):
            tg.create_task(consumer(queue, f"消费者{i+1}"))

asyncio.run(pipeline_demo())

5.2 信号量控制并发度

import asyncio
import aiohttp

async def fetch_url(session: aiohttp.ClientSession,
                    semaphore: asyncio.Semaphore,
                    url: str) -> dict:
    """带并发限制的 HTTP 请求"""
    async with semaphore:  # 同时最多 10 个请求
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                return {
                    "url": url,
                    "status": resp.status,
                    "size": len(await resp.read())
                }
        except Exception as e:
            return {"url": url, "error": str(e)}

async def crawl(urls: list[str], max_concurrent: int = 10):
    semaphore = asyncio.Semaphore(max_concurrent)  # 背压控制
    
    async with aiohttp.ClientSession() as session:
        async with asyncio.TaskGroup() as tg:
            tasks = [
                tg.create_task(fetch_url(session, semaphore, url))
                for url in urls
            ]
    
    results = [t.result() for t in tasks]
    
    success = [r for r in results if "error" not in r]
    failed  = [r for r in results if "error" in r]
    print(f"成功: {len(success)}, 失败: {len(failed)}")
    return results

---

六、线程池与进程池：跨越异步边界

asyncio 的 event loop 是单线程的。遇到 CPU 密集型任务或不支持异步的阻塞 I/O，必须把它们推到线程池或进程池里，否则会卡死整个 event loop。

6.1 边界判断

任务类型                    推荐方案
─────────────────────────────────────────────────────
纯 I/O（网络、文件）        直接用 asyncio 协程
CPU 密集型（计算、压缩）    ProcessPoolExecutor
阻塞 I/O（老旧同步库）      ThreadPoolExecutor
混合型                      按子任务类型分别处理

6.2 run_in_executor 的正确用法

import asyncio
import concurrent.futures
import hashlib
import time

def cpu_intensive_hash(data: bytes, iterations: int) -> str:
    """CPU 密集型：多次哈希计算"""
    result = data
    for _ in range(iterations):
        result = hashlib.sha256(result).digest()
    return result.hex()

def blocking_io_operation(filename: str) -> str:
    """阻塞 I/O：使用不支持异步的老旧库"""
    time.sleep(0.1)  # 模拟阻塞操作
    return f"读取 {filename} 完成"

async def main():
    loop = asyncio.get_event_loop()
    
    # CPU 密集型 → 进程池（绕过 GIL）
    with concurrent.futures.ProcessPoolExecutor(max_workers=4) as process_pool:
        hash_result = await loop.run_in_executor(
            process_pool,
            cpu_intensive_hash,
            b"large data" * 1000,
            10000
        )
        print(f"哈希结果: {hash_result[:16]}...")
    
    # 阻塞 I/O → 线程池
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as thread_pool:
        async with asyncio.TaskGroup() as tg:
            tasks = [
                tg.create_task(
                    loop.run_in_executor(thread_pool, blocking_io_operation, f"file_{i}.txt")
                )
                for i in range(5)
            ]
    
    results = [t.result() for t in tasks]
    print(f"完成 {len(results)} 个文件操作")

asyncio.run(main())

6.3 避免 event loop 被卡死的检测

import asyncio
import time

async def monitor_event_loop(threshold_ms: float = 50):
    """监控 event loop 延迟，检测阻塞操作"""
    while True:
        start = time.perf_counter()
        await asyncio.sleep(0)  # 让出控制权
        elapsed = (time.perf_counter() - start) * 1000
        
        if elapsed > threshold_ms:
            print(f"⚠️  Event loop 延迟: {elapsed:.1f}ms，可能有阻塞操作！")
        
        await asyncio.sleep(0.1)

async def bad_blocking_code():
    """错误示范：在协程里直接调用阻塞操作"""
    time.sleep(2)  # 这会卡死整个 event loop！
    return "done"

async def good_async_code():
    """正确做法：阻塞操作放到线程池"""
    loop = asyncio.get_event_loop()
    with concurrent.futures.ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, time.sleep, 2)
    return "done"

---

七、调试技巧：让异步问题无处遁形

7.1 开启调试模式

import asyncio
import logging

# 开启 asyncio 调试模式
asyncio.run(main(), debug=True)

# 或者通过环境变量
# PYTHONASYNCIODEBUG=1 python your_script.py

# 配置日志，捕获慢协程警告
logging.basicConfig(level=logging.DEBUG)
# asyncio 会自动警告执行时间超过 0.1s 的协程

7.2 任务追踪与可观测性

import asyncio

def task_done_callback(task: asyncio.Task):
    """统一的任务完成回调，用于监控和日志"""
    if task.cancelled():
        print(f"任务 [{task.get_name()}] 被取消")
    elif task.exception():
        exc = task.exception()
        print(f"任务 [{task.get_name()}] 异常: {exc}")
        # 这里可以接入 Sentry、日志系统等
    else:
        print(f"任务 [{task.get_name()}] 完成: {task.result()}")

async def main():
    task = asyncio.create_task(
        some_coroutine(),
        name="重要业务任务"  # 给任务命名，方便调试
    )
    task.add_done_callback(task_done_callback)
    
    # 查看当前所有运行中的任务
    all_tasks = asyncio.all_tasks()
    print(f"当前运行任务数: {len(all_tasks)}")
    for t in all_tasks:
        print(f"  - {t.get_name()}: {t.get_coro()}")
    
    await task

---

八、完整实战：异步任务调度器

把上面所有概念整合成一个可用的任务调度器：

import asyncio
import time
from dataclasses import dataclass, field
from typing import Callable, Any

@dataclass
class TaskResult:
    name: str
    success: bool
    result: Any = None
    error: str = None
    duration: float = 0.0

class AsyncTaskScheduler:
    """
    生产级异步任务调度器
    - 并发控制（背压）
    - 超时管理
    - 错误隔离
    - 结果收集
    """
    
    def __init__(self, max_concurrent: int = 10, default_timeout: float = 30.0):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.default_timeout = default_timeout
        self._results: list[TaskResult] = []
    
    async def _run_task(self, name: str, coro, timeout: float) -> TaskResult:
        start = time.perf_counter()
        async with self.semaphore:  # 背压控制
            try:
                async with asyncio.timeout(timeout):
                    result = await coro
                    return TaskResult(
                        name=name, success=True, result=result,
                        duration=time.perf_counter() - start
                    )
            except TimeoutError:
                return TaskResult(
                    name=name, success=False, error="超时",
                    duration=time.perf_counter() - start
                )
            except asyncio.CancelledError:
                raise  # 取消信号必须传播
            except Exception as e:
                return TaskResult(
                    name=name, success=False, error=str(e),
                    duration=time.perf_counter() - start
                )
    
    async def run_all(self, tasks: list[tuple[str, Any]],
                      timeout: float = None) -> list[TaskResult]:
        """并发执行所有任务，收集结果"""
        timeout = timeout or self.default_timeout
        
        async with asyncio.TaskGroup() as tg:
            scheduled = [
                tg.create_task(
                    self._run_task(name, coro, timeout),
                    name=name
                )
                for name, coro in tasks
            ]
        
        results = [t.result() for t in scheduled]
        
        success_count = sum(1 for r in results if r.success)
        avg_duration = sum(r.duration for r in results) / len(results)
        print(f"完成 {len(results)} 个任务: "
              f"{success_count} 成功, {len(results)-success_count} 失败, "
              f"平均耗时 {avg_duration:.2f}s")
        
        return results

# 使用示例
async def demo():
    async def fetch(url: str) -> str:
        await asyncio.sleep(0.5)
        return f"{url} 响应"
    
    scheduler = AsyncTaskScheduler(max_concurrent=5, default_timeout=2.0)
    
    tasks = [(f"请求{i}", fetch(f"https://api.example.com/{i}"))
             for i in range(20)]
    
    results = await scheduler.run_all(tasks)
    
    for r in results[:3]:
        print(f"{r.name}: {'✓' if r.success else '✗'} ({r.duration:.2f}s)")

asyncio.run(demo())

---

九、总结

asyncio 的核心思想是协作式多任务：每个协程在等待 I/O 时主动让出控制权，让 event loop 去运行其他任务。理解了这一点，很多"奇怪行为"就有了解释。

几个关键原则：

• 用 TaskGroup 替代 gather，结构化并发，杜绝任务泄漏
• CancelledError 必须重新抛出，取消传播不能被吞掉
• 用 Queue + Semaphore 实现背压，保护消费者
• CPU 密集型任务必须进进程池，阻塞 I/O 进线程池
• 给任务命名，开启调试模式，让问题早点暴露

asyncio 的学习曲线确实陡，但一旦建立了正确的心智模型，它会成为你工具箱里最锋利的那把刀。

---

你在 asyncio 实战中踩过哪些坑？ 是取消传播没处理好导致资源泄漏，还是在协程里误用了阻塞调用把 event loop 卡死？欢迎在评论区分享，这类经验比任何文档都有价值。

---

附录：参考资料

资源	链接
asyncio 官方文档	https://docs.python.org/3/library/asyncio.html
TaskGroup PEP 654	https://peps.python.org/pep-0654/
asyncio 调试指南	https://docs.python.org/3/library/asyncio-dev.html
aiohttp 文档	https://docs.aiohttp.org/
Trio（结构化并发参考实现）	https://trio.readthedocs.io/

推荐书籍：《Python 并发编程实战》、《流畅的 Python》第二版第 21 章