Celery 丢任务，多数情况并不是消息被“吃掉”了，而是在几个交接环节上没人接手：消息进入 broker 前没有持久化、Worker 取到消息后进程被杀、任务失败后没有重试或记录、可见性超时把还在执行的任务又投递了一次。要减少丢任务，通常要把“至少投递一次 + 任务可重复执行”当成一个组合来做，而不是只打开某一个开关。先确认三件事：broker 用的是 RabbitMQ 还是 Redis、队列和消息是否持久化、业务任务重复执行一次会不会出问题。第三点决定你能不能使用 late ack。
要让 Celery 尽量不丢任务，通常需要同时做三件事：让 broker 侧持久化生效；把关键任务设为 late ack，并在 Worker 异常退出时把消息放回队列；业务侧把任务写成可重复执行。只改 acks_late 而不做幂等，超时重投会产生重复执行；只做幂等而不改 ack 时机，被 kill 的任务仍然会直接消失。验证方式是杀掉 Worker 进程后看队列是否回补、任务是否重跑，不能只看日志里出现过 SUCCESS。
先确认当前场景和目标
“不丢任务”在不同环境里的含义不一样，先把判断条件列清楚再动手，可以避免改出一堆副作用：
- broker 类型：RabbitMQ 的持久化由队列 durable 和消息 delivery mode 决定；Redis 作为 broker 依赖持久化配置和可见性超时机制，两者语义有差别，处理方式不能互相照搬。
- 任务能否重复执行：发短信、扣款、写流水这类不可重入的操作，必须先有幂等键或去重记录，否则不要急着开 late ack。
- 任务最长执行时间：它直接决定可见性超时该设多大。执行时间超过超时值，消息会被重复投递。
- 失败后要不要人工介入：失败任务直接丢弃，还是落到重试队列或失败记录表，这决定是否需要死信或补投机制。
- 现状配置：先查看现有的
task_acks_late、task_reject_on_worker_lost、worker_prefetch_multiplier是否已被改过，避免叠加冲突。
如果任务是可重复的、允许重跑，就可以采用“至少一次投递 + 幂等”的路线；如果任务本身不可重入，又暂时加不上幂等，只能保守处理：先保证 broker 持久化，接受极端情况下少量丢失，而不是用不完整的 late ack 换来大量重复单。
按这个顺序处理
建议按“先 broker、再 Worker 的 ack 策略、再任务代码、最后是兜底记录”的顺序推进。每一步改动范围小，出问题时也好回退。
第一步：确认 broker 侧真的把消息落盘了
RabbitMQ 需要队列声明为 durable、消息以 persistent 模式投递，否则 broker 重启后队列和消息都会消失。Redis 作为 broker 时，需要开启 AOF，并确认内存淘汰策略不会把队列键驱逐掉；如果淘汰策略配成随机或 LRU 类，队列键可能被清掉，这种丢失在应用日志里看不出来。下面这段配置只是占位示例，连接串和超时值需要按自己的环境替换：
# celeryconfig.py，键名请按自己安装的版本核对
broker_url = "amqp://user:pass@broker-host:5672//"
result_backend = "redis://backend-host:6379/1"
task_default_queue = "default"
task_default_delivery_mode = "persistent" # 持久化投递
broker_transport_options = {
"visibility_timeout": 3600, # 必须大于任务最长执行时间
"max_retries": 3,
}第二步：调整 ack 时机和预取数量
默认情况下，Worker 取到消息后就会 ack，进程一旦被 kill 或机器重启，这条消息就没有了。改成执行完再 ack，并允许 Worker 异常退出时把消息放回队列：
task_acks_late = True # 任务执行完再确认
task_reject_on_worker_lost = True # 子进程被强杀时把消息退回队列
task_acks_on_failure_or_timeout = True # 明确失败也 ack，由重试机制接手
worker_prefetch_multiplier = 1 # 长任务减少一次性预取
worker_max_tasks_per_child = 200 # 定期回收子进程，降低内存泄漏影响prefetch_multiplier 调小之后，单个 Worker 手里暂存的消息变少，进程被杀时“取走但未执行”的消息也相应减少。这里会牺牲一点吞吐，需要结合任务耗时决定。
第三步：把任务写成可重复执行
late ack 的代价是可能重复执行，所以任务本身要有幂等保护。下面的骨架是通用写法，存储部分要换成你自己的数据库或缓存：
@app.task(
bind=True,
acks_late=True,
autoretry_for=(Exception,),
retry_backoff=True,
retry_kwargs={"max_retries": 5},
)
def settle_order(self, order_id):
# 幂等检查：已处理过就直接返回，避免重投造成重复记账
if is_settled(order_id):
return "skipped"
do_settle(order_id) # 业务操作
mark_settled(order_id) # 与业务操作尽量放在同一事务或补偿逻辑里
return "done"注意业务写入和幂等标记之间的顺序：先写标记再执行、或先执行再写标记，都会在异常中断时留下不一致。通常建议用唯一约束或状态机来保证同一笔业务只会生效一次。
第四步：给失败和超时留出口
重试次数耗尽的任务不要只停留在日志里。可以把它写入一张失败任务表，或者路由到一个单独的低优先级队列，方便人工补跑。这一步和“不丢任务”直接相关：没有出口的失败，本质上等同于丢失。
改完后怎么验证
验证要模拟真实故障，而不是看任务有没有正常跑完。可以先用一条能识别的测试任务（例如往一张测试表写一条带唯一标记的记录）走完整条链路：
- 投递任务后，在 broker 侧查看队列积压：RabbitMQ 可以用队列列表命令，Redis 可以用
LLEN查看对应键长度。先确认消息确实进入了队列。 - 在 Worker 执行过程中执行
kill -9杀掉子进程，模拟异常退出。 - 再次查看队列长度和 Worker 的 reserved 状态，预期是消息数量回补或任务重新出现在 reserved 中，而不是就此归零。
- 重启 Worker 后，查看任务是否被重新执行，同时检查测试表里是否只有一条记录，以确认幂等生效。
- 把
visibility_timeout临时设置为小于任务耗时的值，观察超时后是否出现重复投递，以确认这个参数必须按最长耗时来设置。
预期信号是：Worker 被杀不会导致任务凭空消失、任务重新执行但业务结果不重复、失败任务能在失败表或死信队列里查到。若杀进程后队列长度直接归零，说明 ack 时机或 broker 持久化还有一项没生效。
容易忽略的边界
- 不可重入任务：扣款、发券这类操作如果还没做幂等，不要一开始就在全局开启 late ack，可以只对可重复的任务单独声明
acks_late=True。 - 可见性超时：设置得比任务执行时间短，会持续产生重复投递；设置得过大，Worker 崩溃后消息要等很久才回补，恢复时间会变长。需要按任务实际耗时留足余量。
- Redis 作为 broker：可见性机制和 ack 语义与 RabbitMQ 不同，长任务要结合超时值单独评估；内存淘汰策略和持久化配置不当造成的丢失，不在应用层可见。
- 结果后端丢失：result backend 丢数据只影响状态回查，不影响任务执行本身，排查时不要把它和任务丢失混为一谈。
- 监控中的未确认状态：开启 late ack 后，执行中的任务会停留在 unacked 状态，看板上的“积压”数值会偏高，需要区分是正在执行还是真的堆积。
- 定时调度：beat 侧的调度记录和队列任务不是一回事，调度器重启可能造成漏触发，需要单独检查。
- 版本差异：不同大版本的配置项名称和默认值有变化，改之前先按自己安装的版本核对键名，避免写了不生效的配置还以为已经改好。
- 回退建议：如果开启 late ack 后出现大量重复执行，先关掉
task_reject_on_worker_lost止血，再补幂等逻辑；不要用加大 prefetch 或延长超时来掩盖重复问题。