asyncio.Queue 的 join 一直等待：task_done 记账与业务成功分开设计

队列已经空了，await queue.join() 却没有返回，先别继续增加消费者。empty() 只看当前有没有排队项，join() 等的是未完成计数降到零；一个任务可能已经被取走，但对应的 task_done() 永远没有执行。

这份计数也不是业务成功确认。任务校验失败以后，只要已经完成它的处理流程，依然可以结束队列账目；失败结果应保存在另一份业务记录里。把 task_done() 理解成“只对成功任务确认”，会让允许部分失败的批次一直等下去。

下面固定 Python 3.12，做一个有界队列、固定工作者、逐项结果和正常退出信号组成的小型批处理程序。它既能处理预期的输入错误，也不会在工作者意外退出以后，把生产者永远留在 join() 里。

一次 put 增一笔，一次 task_done 销一笔

put() 成功放入一项，未完成计数增加；get() 只是取走待处理项，不减少这份计数。消费者完成对应处理后调用 task_done()，计数才减少。队列长度和未完成计数因此可能在很长一段时间里不同。

可以用一个确定的最小实验观察：

import asyncio


async def accounting():
    queue = asyncio.Queue()
    await queue.put("item")
    assert await queue.get() == "item"
    assert queue.empty()
    try:
        await asyncio.wait_for(queue.join(), timeout=0.05)
    except TimeoutError:
        print("empty, but not finished")
    else:
        raise AssertionError("unfinished item was ignored")
    queue.task_done()
    await queue.join()
    print("accounting finished")


asyncio.run(accounting())

这里超时只用来证明遗漏记账会持续等待，不是在测试处理速度。补上 task_done() 后，join 才能结束。反过来，调用次数超过 put 数量会抛 ValueError，这也不是可以忽略的普通业务错误，而是队列使用协议已经失配。

完整批处理：业务失败保留为结果，队列账目始终配对

保存为 queue_batch.py：

import asyncio

STOP = object()


class RejectedItem(Exception):
    pass


async def calculate(value):
    await asyncio.sleep(0)
    if value < 0:
        raise RejectedItem("negative value")
    return value * value


async def worker(queue, results):
    while True:
        item = await queue.get()
        try:
            if item is STOP:
                return
            ident, value = item
            try:
                result = await calculate(value)
            except RejectedItem as error:
                results[ident] = {"ok": False, "error": str(error)}
            else:
                results[ident] = {"ok": True, "value": result}
        finally:
            queue.task_done()


async def run_batch(values, workers=3, capacity=4):
    if workers <= 0 or capacity <= 0:
        raise ValueError("workers and capacity must be positive")
    queue = asyncio.Queue(maxsize=capacity)
    results = {}
    async with asyncio.TaskGroup() as group:
        for _ in range(workers):
            group.create_task(worker(queue, results))
        for ident, value in enumerate(values):
            await queue.put((ident, value))
        await queue.join()
        for _ in range(workers):
            await queue.put(STOP)
        await queue.join()
    return [results[index] for index in range(len(results))]


async def main():
    results = await run_batch([2, -1, 3, 0])
    assert results == [
        {"ok": True, "value": 4},
        {"ok": False, "error": "negative value"},
        {"ok": True, "value": 9},
        {"ok": True, "value": 0},
    ]
    print(results)
    print("all work items and stop signals were accounted for")


if __name__ == "__main__":
    asyncio.run(main())

运行 python3 queue_batch.py，结果顺序与输入一致，负数被记录为失败项，零仍是合法成功结果。工作者可以按任意顺序完成，最终根据输入编号重新组织结果，不依赖字典插入顺序碰巧与输入相同。

task_done() 放在取得队列项之后的 finally 中。get() 本身被取消时，并没有成功拿到一项，也就不应该额外销账；已经取得项目后，无论成功、预期拒绝还是被取消，当前这笔处理账目都能结束。这里的“结束”只代表工作者不再负责这项，不证明外部副作用一定完成。

停止信号也是通过 put 放进去的队列项，所以同样需要 task_done。每个工作者拿到一个独立信号以后退出；只放一个信号并不会自动广播给全部消费者。

为什么还要用 TaskGroup 包住工作者

只用 create_task() 启动几个消费者，再在主任务里等待 join，存在一条坏路径：某个消费者因编程错误退出，剩余消费者也可能无法继续完成所有项目，主任务却没有收到失败，只看见队列一直没结账。

TaskGroup 将工作者与生产循环放在同一个受管理作用域。工作者抛出未处理异常时，任务组会取消其余子任务，也会中断还在作用域中等待的主任务，最后汇总故障。它不能修复错误，但能避免“消费者已经死掉，调用方还在无期限等成功”的生命周期断裂。

预期的负数拒绝在 worker 内转成结果，其他异常继续向外抛。这条分类需要业务自行确定。若把 Exception 全部转成每项失败，就可能把连接池失效、实现错误等整组不可继续的问题伪装成大量普通坏数据；若一个格式错误就取消整组，又可能违背部分成功的产品要求。

取消时即使 finally 调用了 task_done，也不该对外宣称批次完整成功。上面的函数被意外异常中断时不会返回正常结果列表，调用者会收到任务组错误。结果对象里临时留下了多少项，与整个函数有没有成功完成，是两个层次。

可以把输入中的一项暂时换成字符串，让calculate在比较时抛出未预期的TypeError。调用者应收到任务组异常，而不是返回一个少了项目的正常列表。这个失败样本与负数样本作用不同：负数验证部分成功政策，类型错误验证消费者崩溃时主流程能够结束。测试需要同时保留这两条路径，才不会在以后扩大异常捕获范围时悄悄改变整体政策。

重试还存在一种与记账无关的死锁：所有消费者拿到失败项以后，又都在await queue.put重试项，而队列此刻已经被生产者填满。消费者在等空位，只有消费者能取走项目，却已经没有人继续get。把task_done挪到前面也不会创造新的消费者。重试可以交给独立的受控调度流程，或者让当前消费者在有界次数内直接重做，不再把自己堵在同一个已满队列入口。

如果业务重试确实重新入队，那是一次新的put，必须拥有对应的新记账。原项目完成一次处理与后续尝试完成，并不天然共享同一笔未完成计数；最终业务结果应该按稳定项目编号归并，而不是拿队列内部计数当尝试总数或成功总数。

asyncio.Queue也不是给多个系统线程直接共用的线程安全队列。同步生产者来自其他线程时，需要通过事件循环允许的线程安全入口转交，或使用明确的线程间桥接。不能因为Queue这个名字与标准库同步队列相似，就从任意线程调用它的方法。

有界队列限制了待处理项，不会限制所有资源

maxsize=4 让生产者在队列满时等待，构成一条明确背压。当前同时存在的工作者数量固定，待处理队列也受控，因此不会为每个输入都创建一个 Task。

但这份程序仍把所有结果保存在字典中，输入很多时结果内存会增长；输入 values 如果本身已经是巨大列表，也早已占用内存。有界队列不是整批程序内存有界的证明。需要处理大流时，应让结果写入受控存储，或者由下游持续消费，而不是永远累积。

单项值也可能很大。队列上限以项目数计算，并没有限制每个项目的字节数；传十个几百兆对象仍然可能耗尽内存。项目尺寸、总结果尺寸和处理时长都需要各自预算。

Python 3.12 的 asyncio.Queue 没有内建 shutdown 方法，因此示例使用哨兵约定。后续版本新增队列关闭能力时，也要按相应版本的 join 与立即关闭语义重新评估，不能把旧代码的“所有账已销完”直接等同于新模式下的相同保证。

退出后的结果要能解释

队列 join 返回以后，生产的工作项已经按协议完成处理；第二次 join 则把停止项也结清，任务组退出再确认工作者结束。三个动作各有对象，不能用一句“队列清空了”混成同一个时点。

若需要重试失败项，应使用独立业务编号和结果状态，而不是再次调用 task_done 试图修改队列历史。若外部写入结果未知，应查询或按幂等合同恢复，不能因为消费者已经退出就认定那次写入未发生。

排查等待不结束时，沿 put、get、task_done 与工作者最终状态逐条核对，通常比观察 qsize 更接近根因。队列负责分发和记账，业务成功由结果合同说明，工作者寿命由任务管理负责；这三件事分开以后，批处理才能既结束得了，也解释得清。