Python TaskGroup 子任务失败后怎么收口：ExceptionGroup、取消与清理边界

线上批处理服务最容易在异常时出问题：某个子任务失败后，其他关联任务随即被取消，但队列里的未处理工作既没有完成记录，也找不到可恢复的线索。Python 的 asyncio.TaskGroup 会把同组任务的失败情况统一收口为 ExceptionGroup，而队列的 shutdown() 方法又能把停止接收新请求和排空存量任务两个动作明确拆分。

优雅停机的核心顺序是：先让队列不再接收新任务，再让消费者逐步处理存量任务，最后用 join() 等待每个已取出的任务都调用 task_done()；只有遇到超时或者故障恢复的特殊场景，才考虑启用 immediate=True。

先把失败处理拆成三个边界

常规导出服务一般包含三类协程：负责读取新任务的生产者、从队列取任务执行的消费者，以及负责发出停机信号的控制协程。以前常见的做法是往队列里额外塞几个哨兵对象做终止标记，消费者数量一变，要放的哨兵数量就很容易出错；还有一种做法是直接取消所有消费者，队列里没处理完的待执行项完全没有明确的归属逻辑。

Queue.shutdown() 把状态变更逻辑直接封装在队列本身。调用之后队列就不能再写入新任务，已经阻塞在 put() 操作上的生产者会被直接唤醒，同时收到 QueueShutDown 异常。默认的 immediate=False 配置会保留队列里已经入队的存量任务，让消费者可以继续取出执行。

用 shutdown() 切断生产入口

import asyncio

async def produce(queue: asyncio.Queue, jobs: list[str]) -> None:
    try:
        for job in jobs:
            await queue.put(job)
            print("queued", job)
    except asyncio.QueueShutDown:
        # 控制协程先关闭队列时，生产者正常收尾
        print("producer stopped")

async def consume(queue: asyncio.Queue, worker: str) -> None:
    while True:
        try:
            job = await queue.get()
        except asyncio.QueueShutDown:
            print(worker, "drained")
            return
        try:
            await asyncio.sleep(0.02)  # 代表实际处理与确认
            print(worker, "done", job)
        finally:
            queue.task_done()

这里的 finally 处理非常关键。只要消费者已经成功取出一个任务，不管后续处理成功、抛出业务异常还是协程被取消，都要先判断这个任务的最终状态，再调用 task_done()。如果直接从异常分支跳出没有减少未完成计数，后面的 join() 会一直阻塞等待。

排空阶段不要误用 immediate=True

async def run_batch() -> None:
    queue = asyncio.Queue(maxsize=20)
    async with asyncio.TaskGroup() as group:
        group.create_task(produce(queue, ["a", "b", "c", "d"]))
        group.create_task(consume(queue, "worker-1"))
        group.create_task(consume(queue, "worker-2"))

        # 实际服务中，这里由信号处理器或控制面触发
        await asyncio.sleep(0.05)
        queue.shutdown()
        await queue.join()

asyncio.run(run_batch())

常规关闭的语义就是“停止接收新任务，但允许已经入队的任务全部执行完成”。当每个入队任务都配对调用了 task_done() 之后，join() 才会解除阻塞。消费者下一次调用 get() 时，如果队列已经排空，就会收到 QueueShutDown 异常直接退出。

示例里的等待逻辑只是为了模拟控制面收到停机信号的过程；生产环境的关闭流程应该由系统信号、部署控制器或者应用内部的管理接口触发，不要把固定延时当作任务完成的判断条件。完成判断直接交给 join() 就行，它本身就是靠未完成任务计数做判断的。

强制关闭是故障路径，不是快捷键

queue.shutdown(immediate=True) 会立即清空队列，同时唤醒所有当前正在阻塞的 get() 和 join() 调用。这个逻辑适合进程即将被强制杀死、外部依赖已经完全不可用等极端场景，但它可能让 join() 在所有存量任务都没真正处理完成时就返回，不能把这个返回结果当成所有业务都执行成功的标志。

如果必须执行强停，建议同步记录当前的未完成任务数和对应的恢复批次号：已经被消费者取出的任务仍要正常做取消后的清理工作，留在队列里的未处理任务则直接写入可重放的持久化存储。不然服务虽然正常退出了，数据却没有留下重新执行的入口。

TaskGroup 负责收口，队列负责生命周期

asyncio.TaskGroup 很适合用来管理生产者和消费者的协程边界，但它不会替队列决定什么时候停止接收新任务。队列关闭之后，消费者通过捕获 QueueShutDown 自然退出，TaskGroup 在上下文退出时会自动等待所有子协程完成；如果某个任务抛出未处理异常，TaskGroup 会自动取消同组的其他任务，业务代码仍需要在清理路径里维护好外部任务的状态。

常见问题与检查顺序

为什么 join() 一直不返回？

先排查每次成功执行 get() 之后，是不是都有对应的 task_done() 调用，再检查消费者是不是在捕获异常后直接跳出循环，漏掉了计数更新。不要一上来就加 immediate=True ，那只是把计数错误的问题掩盖住了。

为什么停机后还有新任务进入队列？

确认所有生产入口用的是同一个 Queue 实例，同时所有写入操作的 put() 逻辑都加了 QueueShutDown 异常捕获。如果业务还有外部消息拉取器，关闭队列之前也要先暂停它的拉取循环。

什么时候应该保留未完成任务？

只要任务支持重试或者后续补偿，就优先用默认关闭逻辑，执行完所有存量任务；只有进程已经无法继续运行的时候才走强制关闭流程，同时把剩余任务交给持久化的恢复流程处理。

上线前的停机验收清单

先压入一批可观测的测试任务，触发关闭后确认没有新的 put() 操作能写入成功；再确认所有已经取出的任务都打印了完成日志或者进入了补偿记录；最后检查 join() 的解除时间和消费者打出的 QueueShutDown 日志。把正常关闭和强制关闭拆成两个独立的测试用例，不要只校验“进程最终退出”这一层结果。

这样设计之后，整个停机逻辑就不再依赖哨兵对象的数量或者取消操作的时机：shutdown() 管住新任务入口，get() 和 QueueShutDown 管住消费者的退出逻辑，task_done() 与 join() 管住任务完成的校验逻辑。