用户一次提交五个计算任务
假设一个业务系统允许用户选择五个项目，计算企业资质与各项目要求的匹配度。每个项目平均需要五秒，计算过程中可能读取数据库，也可能调用模型服务。
如果接口依次执行五次计算，用户大约要等待二十五秒才能拿到全部结果。一个自然的改进是：接口先返回任务编号，计算在后台进行，前端定期查询状态。
用户提交五个项目
-> 接口返回任务编号
-> 后台继续计算
-> 前端查询进度和结果
做到这里，我们已经有了异步交互。但要判断是否使用了任务队列，还需要沿着任务编号往后看：这些任务放在哪里等待？谁决定开始执行？执行进程退出后，任务还能找到吗？
本文适合已经接触 async、await，正在把耗时操作搬到后台的后端开发者。我们先从等待和执行的关系理解队列，再讨论可靠性和工具选择。
先让接口返回，还缺少什么
一种实现是在 API 进程中调用 asyncio.create_task()，启动计算协程，然后返回响应。前端查询的状态可以存入内存，也可以写入数据库。
API 进程
-> 创建五个计算协程
-> 返回响应
-> 这些协程由当前进程的事件循环调度
这解决了接口等待的问题，但单独调用 create_task() 并没有建立一套业务任务的入队、领取和重新分配机制。如果进程退出，协程就结束了；即使数据库里还保留着 running 状态，也需要另一个执行者找到并恢复任务。
前端轮询只负责观察状态，无法提供这种恢复能力。线程池、进程池或者框架内部可能已经有调度队列，但它们也不因此自动具备业务任务的持久化与恢复语义。
接下来需要解决的问题是：允许任务先等待，再由有空闲能力的执行者领取。
加入一个等待区，生产和消费就可以分开
我们把接口负责的事情限制为“校验请求并提交任务”，把后台执行者负责的事情限制为“领取任务并执行”。两者通过一个保存待处理任务的队列连接。
生产者：接口或其他业务流程
-> 提交任务到队列
消费者：后台 worker
-> 有空闲时领取任务
-> 执行计算
-> 记录结果和完成状态
这里的生产者和消费者是两种职责，不一定是两个进程。它们可以是同一进程内的两个协程，也可以运行在不同机器上。让 API 和 worker 使用独立进程，是常见的部署选择。
现在，用户提交的五个任务可以在没有消费者运行时先留在队列里。消费者启动后再领取；如果有两个执行槽位，就同时处理两个任务，其余三个继续等待。这就是生产消费解耦。
一次请求也不必对应一个 worker。许多用户提交的任务可以进入同一个队列，由一组 worker 共同处理。是否将五个项目拆成五条任务，取决于我们是否需要独立查询、重试和控制每个项目的计算。
用 Python 观察这个过程
Python 标准库的 asyncio.Queue 就能演示这种关系。下面先放入五个任务，再启动两个消费者。这样可以直接看到：提交任务的时候，消费者还没有开始工作。
每个消费者通过 queue.get() 领取一个任务，完成后再领取下一个。queue.join() 等待已入队项目全部调用 task_done()；最后为每个消费者放入一个 None，通知它退出。
下面是完整演示，使用 Python 3.10 或更高版本即可运行，无需安装依赖：
import asyncio
async def worker(
name: str,
queue: asyncio.Queue[int | None],
statuses: dict[int, str],
) -> None:
while True:
task_id = await queue.get()
try:
if task_id is None:
return
statuses[task_id] = "running"
print(f"{name}: start task {task_id}")
await asyncio.sleep(0.2)
statuses[task_id] = "succeeded"
print(f"{name}: finish task {task_id}")
finally:
queue.task_done()
async def main() -> None:
queue: asyncio.Queue[int | None] = asyncio.Queue(maxsize=5)
statuses: dict[int, str] = {}
for task_id in range(1, 6):
queue.put_nowait(task_id)
statuses[task_id] = "pending"
print("accepted:", statuses)
workers = [
asyncio.create_task(worker(f"worker-{number}", queue, statuses))
for number in range(1, 3)
]
await queue.join()
for _ in workers:
await queue.put(None)
await asyncio.gather(*workers)
print("finished:", statuses)
if __name__ == "__main__":
asyncio.run(main())
开头的 accepted 会显示五个 pending。随后可以看到两个 worker 分别开始任务，最终五个状态都变成 succeeded。
这个例子的执行上限是两个任务，因为我们只启动了两个逐条处理任务的消费者。它们在等待期间并发推进，但 sleep() 只是模拟等待，不代表 CPU 计算获得了多核加速。
这已经是一个队列，只是队列和状态都在内存中。进程退出后，尚未完成的任务不会保留下来。另外，task_done() 只是内存队列的完成计数，不能等同于持久化系统中的成功确认。演示中的任务没有失败分支，生产系统还需要明确失败后是重试、记录失败，还是等待人工处理。
队列如何帮助并发，又为什么会积压
继续使用开头每个任务五秒的假设。如果系统允许两个任务同时执行，理想消费能力约为：
每分钟完成任务数 ≈ 并发执行槽位数 × 60 / 单任务平均耗时秒数
= 2 × 60 / 5
= 24
假设每分钟接收三十个任务，而实际只能完成二十四个，那么未完成任务会以每分钟约六个的速度增加。五分钟后就多了约三十个未完成任务。
为了避免把“等待数”和“未完成数”混在一起，可以这样记账。在没有取消和丢弃的前提下：
未完成任务数 = 已接收任务数 - 已完成任务数
未完成任务数 = 等待领取的任务数 + 正在执行的任务数
队列把突发压力缓冲下来，消费者按下游能够承受的速度处理。若长期到达速率 λ 高于实际完成速率 μ，积压就会继续增长。即使两者的平均值相等，任务耗时波动也可能造成较长等待，因此通常需要留出消费余量。
增加 worker 可以提高并发度，但只有在下游仍有余量时才可能提高吞吐。例如模型服务只能有效并行两个请求，启动二十个 worker 很可能只是把等待从任务队列转移到模型服务，甚至增加超时。
生产者和消费者都需要边界
生产者可以在消费者停止期间入队，但不能无限提交。上面的内存队列最多保存五个等待项目；队列满时，put_nowait() 会抛出 QueueFull，而 await queue.put() 会等待空间出现。这种让上游减速的机制叫背压。
业务接口也需要自己的限制，例如一次最多提交十个项目、同一用户限制未完成任务数，并对重复请求去重。队列无法接收任务时，接口应返回明确的拒绝或稍后重试信息。
消费者可以按空闲能力、优先级或定时规则决定何时领取，但长期不消费就会增加等待时间。高优先级任务持续插队，还可能让低优先级任务一直等不到执行机会。
前端因此需要分清两件事：提交成功表示任务已被接受；running 才表示计算已经开始。生产系统应在确认任务可靠入队后，才向用户返回“已受理”。
用数据库任务表支持重启恢复
如果要求 API 重启后任务仍然存在，可以用 PostgreSQL 的任务表保存队列状态。例如每条记录包含任务编号、任务类型、业务对象编号，以及 pending、running、succeeded、failed 等状态。
API 提交任务并提交事务
-> 任务表留下 pending 记录
-> worker 在短事务中锁定并领取
-> 提交领取事务
-> 在事务之外执行耗时计算
-> 在新的短事务中保存结果和完成状态
把任务写进表还不够，还需要领取规则。多个 worker 查询同一批待处理任务时，可以使用 FOR UPDATE SKIP LOCKED：一个领取事务锁定某行后，另一个领取事务跳过这行，继续找其他任务。
这种行锁保护的是领取过程。事务提交后锁就释放了，不能让数据库事务一直覆盖几分钟的模型调用。执行期间的任务所有权需要另一种机制。
用租约回答“worker 消失了怎么办”
假设 worker A 领取任务时获得十分钟租约，并每分钟续租。任务记录可以包含以下信息：
| 字段 | 作用 |
|---|---|
worker_id | 当前由哪个执行进程持有任务 |
lease_expires_at | 当前所有权何时过期 |
lease_version | 每次重新领取时递增的所有权版本 |
heartbeat_at | 最近一次成功续租时间，便于监控 |
这十分钟和一分钟只是一个例子，实际值要结合任务时长、续租延迟和故障恢复要求选择。
一次中断的时间线可能是：
00:00 A 领取任务，版本 7，租约到 00:10
00:01 A 续租成功，租约延长到 00:11
00:02 A 退出，心跳停止
00:11 租约到期
随后 B 轮询并接管，版本递增到 8
接管发生在另一个 worker 领取任务的时候。它不需要把所有 running 任务先重置为 pending，只应接管符合过期条件的任务。
如果 A 只是网络中断或短暂暂停，它可能在 B 接管后继续运行。因此 A 保存结果时，必须携带原先领取的 worker 标识和版本，并校验状态、所有者、版本及租约仍有效。版本已经是 8，A 持有的版本 7 就不能发布结果。这种阻止旧持有者写回的校验通常称为 fencing。
当计算结果和任务记录都在同一个数据库中时，所有权校验、结果写入和完成状态应在同一事务中提交。只在模型调用之前检查一次，无法阻止调用过程中发生接管后的旧结果写回。
同一个任务仍然可能执行多次
租约允许恢复，也意味着任务可能被重新执行。即使过期结果不能写入数据库，旧 worker 已经发出的外部模型请求，也不一定会随着租约失效立即停止。
更麻烦的是发送通知或扣费：外部调用已经成功，但进程在记录成功前退出，重新执行时可能再次调用。因此需要为可重复执行的业务定义幂等键，或者利用下游的幂等接口。数据库内的 fencing 并不能撤销已经发生的外部副作用。
这里能建立的保证是：只有当前有效持有者可以提交受保护的结果。它不等于“计算函数永远只调用一次”。
租约过期接管和业务失败重试也要分开。模型返回错误后，任务可以明确结束为 failed；是否重试、间隔多久、最多几次，需要另行定义，不能假设有租约就会自动重试所有错误。
Celery、Redis 和 RabbitMQ 分别负责什么
理解生产、等待、领取和完成以后，再看这些工具就容易得多。它们没有全都承担同一种职责。
| 组件 | 所在层次 | 主要责任 |
|---|---|---|
| Celery | Python 任务框架 | 提交任务、运行 worker、配置并发、重试和超时等执行行为 |
| RabbitMQ | 消息代理 | 路由消息、保存队列、投递消息和处理消费者确认 |
| Redis | 数据存储及相关通信机制 | 通过 List、Stream 等结构承载排队和消费机制，也可用于结果存储 |
| PostgreSQL 任务表 | 应用自行组织的数据库队列 | 持久化任务；应用实现领取、租约、状态和恢复规则 |
Celery 的常见部署关系是：
API 中的 Celery 客户端
-> Redis 或 RabbitMQ 作为 broker
-> Celery worker 执行任务
-> 将结果写入业务数据库或可选的 result backend
Broker 负责传递任务消息，result backend 用来保存任务结果或查询元数据，两者可以使用同一种产品，也可以使用不同产品。是否使用 Celery 的结果后端，取决于应用需要；业务数据库里的计算结果并不必然交给它管理。
Celery 可以减少自己维护执行框架的工作，但可靠性仍然依赖消息持久化、确认时机、故障后的重新投递策略和业务幂等设计。例如处理前确认与处理后确认，对进程中断后的行为就有不同影响。
Redis List：最直接的入队与取出
一个简单方案是生产者使用 LPUSH 写入列表，消费者使用 BRPOP 从另一端阻塞取出。消费者没有任务时等待，有任务时继续处理。
问题是：取出任务后，如果消费者在处理完成前退出，这条任务已经离开原列表。单靠这两个命令，无法知道它是否处理成功。
可以用 BLMOVE 原子地把任务从待处理列表移到处理中列表，成功后再从处理中列表移除。但超时恢复、执行归属和重复处理仍需要应用定义。这就是最初几个 Redis 命令之外的工程成本。
Redis Streams：跟踪尚未确认的消息
Streams 提供消费者组，可以让组内多个消费者共同领取消息：
- 生产者用
XADD添加消息。 - 消费者用
XREADGROUP读取，已交付但尚未确认的消息会被跟踪在 pending entries list 中。 - 处理完成后用
XACK确认。 - 应用可以用
XAUTOCLAIM重新认领空闲时间超过阈值的未确认消息。
这里的“pending”表示未确认，可能正在处理，不能直接理解为还没开始。XACK 也不等于从 Stream 删除消息，保留和清理要另行安排。
Streams 提供了恢复所需的工具，但需要应用主动执行认领逻辑；它也不能判断业务副作用是否已经发生。Redis 的持久化、复制与故障切换配置同样影响数据安全。
Celery 的 Redis broker 使用其自身的传输实现，不能因为选择了 Redis，就认为它自动变成了一套由我们定义的 Streams 消费者组。
RabbitMQ：由代理管理投递和确认
RabbitMQ 的工作队列允许多个消费者共同处理任务。使用手动确认时，消费者处理成功后发送 ACK；连接关闭时，未确认的消息通常可以重新投递。
消息持久化、队列类型、生产者发布确认和消费者确认共同决定故障行为。只配置“有一个 RabbitMQ 队列”，也不能直接得到业务恰好执行一次的保证。
Kafka 更偏向可保留、可回放的事件流，适合多个消费组独立读取同一批事件。它也能承载任务分发，但普通后台计算不必仅仅为了使用队列，就先引入 Kafka。
从现有业务需求决定下一步
对于已有 PostgreSQL、任务规模适中、结果也保存在同一数据库里的系统，任务表是一种合理起点。它方便查询，也便于把业务状态和任务入队放在同一个事务里。
如果后来任务类型、消费者规模或跨服务通信明显增加，再根据实测的数据库轮询压力、排队时间、恢复要求和运维能力，评估 Celery 配合 Redis 或 RabbitMQ。没有一个单独的任务数量，能替所有系统决定迁移阈值。
判断自己的后台计算方式时，可以沿着一次提交检查：
- 任务被接受后在哪里等待，进程退出后还能否找到？
- 谁负责领取，允许同时执行多少条？
- 任务成功、失败或中断后，状态由谁更新？
- 重新执行时，怎样避免重复提交结果或重复产生副作用？
对于开头的五个项目，前端无论采用轮询还是推送，都只是在观察计算进度。真正决定后台行为的是任务的入队、领取、执行和恢复协议。
可以先运行 Python 演示，再把消费者数量从二改成一，观察等待和完成顺序。随后对照自己的服务重启流程，确认它能恢复哪些任务，以及用户看到的“已受理”究竟承诺了什么。