这是 Beta 探索课程，内容结构、实验步骤和示例可能会继续调整。

可靠性

那次惨痛的教训

2024 年 4 月 1 日（愚人节，但对我们来说不是玩笑），我们的延时队列服务发生了一起严重事故。

事故经过

时间线：

14:00 - 用户 A 下单，创建 30 分钟后取消订单的任务
14:05 - 用户 A 支付成功，取消超时任务
14:30 - 延时队列服务重启（部署新版本）
14:31 - 用户 A 收到订单被取消的通知 🤯

根因分析

经过排查，问题出在这里：

问题根源：

❌ 任务只存储在内存，无持久化
❌ 服务重启后，任务全部丢失
❌ 取消任务只是从内存删除，没有记录
❌ 重启后，任务不存在，但用户的期望还在

可靠性的三个维度

1. 任务不丢失（No Loss）

任务创建后，无论发生什么情况（服务器崩溃、网络故障、停电），都不能丢失。

2. 任务不重复（No Duplicate）

同一个任务不能被多次执行。如果执行失败需要重试，要有明确的控制。

3. 任务不遗漏（No Miss）

到期的任务必须被执行，不能因为任何原因被遗漏。

可靠性三角
    不丢失
      / \
     /   \
    /_____\
   不遗漏 不重复

任务不丢失的保障

方案 1：双层存储（数据库 + 内存）

双层存储的优势：

层级	存储	优点	缺点
L1: 数据库	磁盘	✅ 强持久化 ✅ 事务保证 ✅ 可靠性高	⚠️ 性能较低
L2: Redis	内存	✅ 高性能 ✅ 快速查询	⚠️ 需要持久化配置

方案 2：WAL（Write-Ahead Log）

使用写前日志确保数据不丢失：

WAL 的思路是：先把”我要做什么”记进日志，再去做。只要日志落盘了，哪怕进程在真正执行前崩溃，重启后也能照着日志把没做完的事补上。

1. 收到「30 分钟后取消订单 ORD001」的任务
2. 把任务内容 + 执行时间追加写入 WAL 文件，并 fsync
3. fsync 返回成功后，才向调用方返回「已接收」
4. 后台线程读取 WAL，把任务投递到真正的延时存储

关键在于第 2 步和第 3 步的顺序不能反。如果先返回成功再落盘，那么”返回成功”和”任务真的存在”之间就有一个窗口期——进程恰好在窗口期内崩溃，任务就凭空消失了，而这正是最初那次事故的形态。

WAL 的代价是：

每次写都要 fsync，磁盘 IO 延迟直接叠加到任务创建的响应时间里；
日志文件会持续增长，需要定期做快照（checkpoint）并截断；
日志损坏时可能丢失最后一段，所以还需要校验和与日志轮转。

所以 WAL 通常不单独使用，而是配合方案 1 的双层存储一起用：内存里放”等待中的任务”（快），磁盘上放 WAL（可恢复），数据库做最终兜底（权威）。

方案 3：消息队列持久化

使用 Kafka 等持久化消息队列：

这类方案的思路是把”等待”这件事交给消息队列本身——延迟消息是队列的一个原生能力，任务在队列内部按时间排队，到期才被投递出来。

生产者 ──发送延时消息──► [ Kafka / RocketMQ ]
                              │
                              │ 到期投递
                              ▼
                           消费者

它的优势很直接：

持久化由队列负责，消息落盘后才向生产者确认，进程崩溃不会丢消息；
消费语义成熟，至少一次、幂等消费、死信队列这些机制是现成的；
天然分布式，多个消费者组可以分担到期任务；
水平扩展，分区数决定了并发消费能力。

代价也必须说清楚：

需要引入并运维一个中间件，多了一个会挂的组件；
延时能力依赖具体实现。Kafka 本身没有原生延时消息（要靠多级 topic 或延迟插件），RocketMQ 和 RabbitMQ 的延时能力则各有不同的限制；
投递精度取决于队列的调度，不如数据库轮询那样可以精确到秒级可控；
消费失败要自己处理重试与死信，否则任务会卡在某个状态。

小结：三种方案不是互斥的。生产上常见的组合是——Redis ZSet 做热路径的延时存储 + 数据库做权威记录 + MQ 做跨进程投递 + 定时对账兜底。选哪一种，取决于你能接受多大的组件复杂度和多高的数据丢失容忍度。

任务不重复的保障

幂等性设计

确保同一个任务即使被多次调用，结果也是一样的。

幂等不是”保证不重复”，而是”重复了也没关系”。在分布式系统里，重复投递几乎必然发生——网络超时后重发、消费者处理成功但 ACK 丢失导致重投、多副本同时取到同一条记录。所以正确的目标不是”杜绝重复”，而是”让重复执行产生相同的结果”。

最常见的做法是幂等键 + 唯一约束：

-- 每个任务带一个全局唯一的幂等键
INSERT INTO task_log (idempotent_key, task_id, result)
VALUES ('ORD20241015001:cancel', 'TASK001', 'OK');

如果同一个任务被执行了两次，第二次的插入会撞上唯一索引报错，业务代码据此判断”这个任务我已经处理过了”，直接返回上次的结果，不再重复执行副作用。

三层幂等检查

光有数据库唯一约束还不够——它只能保护”已经落库的那部分副作用”。三层检查把幂等保护铺到了整条链路上：

第一层：入口去重（防重复入队）

同一个任务不应被排进队列两次。做法是给每个任务算一个业务幂等键（如 订单号 + 动作类型），入队前先查是否已存在。

key = f"{order_id}:cancel"
if not redis.set(key, "1", nx=True, ex=3600):
    return  # 已在队列里，直接返回
queue.schedule(payload, delay=timedelta(minutes=30))

这里用 Redis 的 SET NX 是因为它既原子又快；而用数据库唯一索引也可以，代价是多一次写。

第二层：消费去重（防重复执行）

任务从队列取出后，在真正执行副作用之前，先用幂等键抢占一个”处理中”标记。抢到才执行，抢不到说明别人正在处理或已经处理完，直接跳过。

if not redis.set(f"lock:{task.key}", "1", nx=True, ex=60):
    return  # 正在处理，直接跳过
try:
    do_business_side_effect(task)
    mark_task_done(task.id)
finally:
    redis.delete(f"lock:{task.key}")

第三层：结果兜底（防重复落库）

前两层都是”尽力而为”的拦截，最终的防线是数据库。副作用落库时带上幂等键和唯一索引，即使前两层都被绕过（比如 Redis 挂了、锁超时了），数据库也不会产生两条记录。

INSERT INTO order_log (idempotent_key, order_id, action)
VALUES (:key, :order_id, :action);
-- 唯一索引冲突 → 视为已处理，捕获异常并返回成功

三层是逐层收紧的关系：入口层挡住重复排队，消费层挡住并发重复执行，兜底层挡住前两层全部失效时的重复落库。缺任何一层都会留下漏洞——最典型的就是只做第三层，导致重复消费时业务逻辑（发券、扣款、推送）被执行了多次，只是最后写库时被唯一索引拦下。

任务不遗漏的保障

补偿机制

定期检查是否有遗漏的任务：

补偿机制的思路是不指望主链路一次成功，而是用一条独立的检查链路兜住漏网之鱼。

-- 每 5 分钟跑一次：找出已经过期但还停在 waiting 状态的任务
SELECT id, task_type, payload, expire_at
FROM delayed_task
WHERE status = 'waiting'
  AND expire_at <= NOW() - INTERVAL 5 MINUTE
LIMIT 1000;

查到就重新投递到队列，并把 retry_count 加一。连续重试超过阈值（比如 10 次）仍然失败的任务，转入死信表并告警——宁可让任务显式地失败在死信里，也不要让它静默地消失。

补偿任务需要满足三个条件才能真正兜底：

查询走索引，否则任务量大时补偿任务本身会变成新的慢查询；
有去重保护，补偿和主链路可能同时处理同一个任务，靠幂等键避免重复执行；
有积压告警，补偿队列长期堆积说明主链路出问题了，这个信号比单个任务失败重要得多。

心跳机制

消费者定期发送心跳，证明自己还活着：

心跳解决的是另一个问题：任务被取走了，但消费者在执行过程中挂了，任务就永远卡在”处理中”。补偿机制能发现”过期还在 waiting”的任务，却发现不了”已经取走但没做完”的任务——因为对账时它看起来是处理中，不是遗漏。

心跳方案给每个消费者一个租约：

# 消费者每 10 秒刷新一次租约，租约有效期 30 秒
def heartbeat(consumer_id):
    redis.setex(f"lease:{consumer_id}", 30, now())

def scan_stuck_tasks():
    now_ts = now()
    for task in db.find_processing_tasks():
        last_beat = redis.get(f"lease:{task.consumer_id}")
        if not last_beat or now_ts - float(last_beat) > 30:
            # 消费者已死：任务退回 waiting，并累加重试次数
            db.reset_to_waiting(task.id)
            db.incr_retry(task.id)