开发者日记

外观

告警不丢：BRPOPLPUSH 安全队列实战 ​

问题背景：告警风暴下，为什么不能直接消费 ​

监控系统在故障发生时，往往不是产生一条告警，而是同一服务、同一机房、同一根因在几十秒内连续触发数百条通知。直接把每条告警发给值班人，会造成通知风暴、重复工单和关键信息被淹没。

一个常见做法是引入 Redis List 做告警收敛队列：采集端把事件 LPUSH 到队列，消费端按服务、指纹或时间窗口聚合同类告警，再决定发送、升级或抑制。但这里有一个容易被忽略的问题：如果消费端使用 LPOP 取出消息后进程崩溃，这条消息已经从 Redis 中消失，告警就可能永久丢失。

因此，生产环境更推荐使用 BRPOPLPUSH：它会以阻塞方式从源队列尾部取出一条消息，并原子写入处理中队列。只有业务处理完成后，才显式从处理中队列删除。这样即使消费者宕机，消息仍然保留在 processing list 中，可由巡检任务回溯或重新入队。

方案对比：List、Lua 与 Stream 怎么选 ​

方案一：LPOP 简单队列 ​

LPOP 的优点是简单、延迟低，适合日志、埋点等允许少量丢失的场景；缺点是取出即删除，没有处理中状态，也没有内置重试次数。告警属于运维可靠性链路，不建议只用 LPOP。

方案二：BRPOPLPUSH 安全队列 ​

BRPOPLPUSH 是 List 结构下非常经典的可靠队列模式。它把“取消息”和“备份到处理中队列”合并成一个原子操作，消费者阻塞等待，避免轮询消耗 CPU。配合 LREM、LREM 回收和独立元数据，可以实现处理中备份、失败回溯、超时重投。

需要注意，Redis 6.2 之后更推荐使用 BLMOVE，BRPOPLPUSH 仍可兼容旧版本；本文命令对大多数线上 Redis 版本都可直接演示。

方案三：Stream 消费组 ​

Redis Stream 支持消费组、待确认列表（PEL）、消息 ID 和 XACK，更适合复杂事件流。但如果当前需求只是单消费者或少量消费者的告警收敛，引入 Stream 会增加学习成本和清理成本。List 方案轻量、直观，便于与现有告警网关集成；当需要多消费组、消息回放、范围查询时，再升级 Stream 更合适。

告警事件采集LPUSH 入队待收敛队列alert:pending处理中队列alert:processing收敛与发送成功后 LREM失败回溯集合alert:dead超时回收任务重新入队

关键数据结构设计 ​

建议至少准备四类 key：

  • alert:pending：List，等待收敛的原始告警。
  • alert:processing:{consumer}：List，某个消费者已取出但尚未确认的告警。
  • alert:meta:{id}：String，保存消息体、开始处理时间、重试次数等 JSON 元数据。
  • alert:dead：List，超过最大重试次数后进入死信队列，用于人工回溯。

List 中不要只放一段很长的 JSON，然后在失败时靠字符串匹配删除。更稳妥的方式是队列里放唯一告警 ID，消息内容写入 alert:meta:{id}。这样 LREM 删除准确，回溯时也能通过 ID 查询完整上下文。

实操步骤：可运行的 BRPOPLPUSH 示例 ​

以下示例可以直接在 redis-cli 中执行。先清空演示 key，再模拟三条同类告警入队。

bash
# 清理演示数据
DEL alert:pending alert:processing:worker-1 alert:dead

# 模拟三条 CPU 告警入队；队列元素只放唯一 ID
LPUSH alert:pending evt:1001 evt:1002 evt:1003

# 写入每条告警的完整内容
SET alert:meta:evt:1001 '{"host":"db-1","rule":"cpu_high","severity":"warning","fingerprint":"db-1_cpu_high"}'
SET alert:meta:evt:1002 '{"host":"db-1","rule":"cpu_high","severity":"warning","fingerprint":"db-1_cpu_high"}'
SET alert:meta:evt:1003 '{"host":"db-1","rule":"cpu_high","severity":"critical","fingerprint":"db-1_cpu_high"}'

# 查看待处理队列长度
LLEN alert:pending

消费者使用 BRPOPLPUSH 阻塞等待消息。超时时间设置为 5 秒，便于命令行演示；线上服务可以设置为 0 持续阻塞，或设置较短时间以便定期检查停机信号。

bash
# 从 pending 尾部取出一条，并原子写入 worker-1 的 processing 队列
# 返回值就是被移动的告警 ID
BRPOPLPUSH alert:pending alert:processing:worker-1 5

# 处理中队列现在应有一条 evt:1003
LRANGE alert:processing:worker-1 0 -1

# 根据 ID 获取完整消息体
GET alert:meta:evt:1003

如果收敛、去重、通知发送都成功，再确认删除。注意：BRPOPLPUSH 每次把新元素推入目标列表的头部，因此处理中队列头部就是当前正在处理的消息。

bash
# 业务处理成功后，从处理中队列删除 1 个匹配元素
LREM alert:processing:worker-1 1 evt:1003

# 元数据可按保留周期删除；若需要审计，可保留一段时间
DEL alert:meta:evt:1003

# 确认处理中队列已空
LRANGE alert:processing:worker-1 0 -1

如果处理失败，不要无限立即重试，否则一个坏消息可能形成热循环。可以用单独 Hash 记录次数，超过阈值后移动到死信队列。

bash
# 演示处理 evt:1002 失败一次
HINCRBY alert:retry evt:1002 1

# 查看当前重试次数
HGET alert:retry evt:1002

# 若未超过阈值，例如 3 次，可重新放回待处理队列
RPUSH alert:pending evt:1002
LREM alert:processing:worker-1 1 evt:1002

# 若超过阈值，转入死信队列，保留现场供回溯
RPUSH alert:dead evt:1002
LREM alert:processing:worker-1 1 evt:1002

# 查看死信队列
LRANGE alert:dead 0 -1
GET alert:meta:evt:1002

消费者伪代码：把原子操作和业务确认分开 ​

下面伪代码展示安全队列的核心边界：Redis 只保证移动原子性，业务成功与否必须由程序显式确认。

bash
# 伪代码，不是 redis-cli 原生命令
# while true:
#   id = BRPOPLPUSH alert:pending alert:processing:worker-1 5
#   if id is null:
#     continue
#   payload = GET alert:meta:{id}
#   try:
#     aggregate_by_fingerprint(payload)
#     send_notification_if_needed(payload)
#     LREM alert:processing:worker-1 1 id
#   except TransientError:
#     retry_count = HINCRBY alert:retry id 1
#     if retry_count < 3:
#       RPUSH alert:pending id
#     else:
#       RPUSH alert:dead id
#     LREM alert:processing:worker-1 1 id
#   except FatalError:
#     RPUSH alert:dead id
#     LREM alert:processing:worker-1 1 id

收敛逻辑可以按 fingerprint 维护时间窗口，例如 5 分钟内同指纹只通知一次，后续事件累加计数并在恢复时发送汇总。队列负责可靠传递，收敛窗口负责降噪，两者不要混为一个超大 List。

超时回收：消费者宕机后如何回溯 ​

仅有 processing 队列还不够，还需要一个回收器扫描长时间未确认的消息。List 本身不支持按成员读取时间，因此实践中有两种做法：

  1. 队列元素使用 ID，并在元数据中写入 started_at，回收器遍历 processing 后读取元数据判断。
  2. 额外维护 Sorted Set，score 为处理开始时间戳，member 为告警 ID。

第二种方式查询超时消息更高效。消费者移动消息成功后，执行 ZADD alert:processing:time {now} {id}；回收器通过 ZRANGEBYSCORE 找出超时 ID，再决定重投或死信处理。

bash
# 假设当前时间为 1710000000，记录处理开始时间
ZADD alert:processing:time 1710000000 evt:1001

# 回收器查找处理超过 300 秒的消息
ZRANGEBYSCORE alert:processing:time -inf 1709999700

# 重投前先从 processing 和时间索引中移除，再放回 pending
LREM alert:processing:worker-1 1 evt:1001
ZREM alert:processing:time evt:1001
RPUSH alert:pending evt:1001

回收器要防止多个实例同时搬运同一条消息。简单场景可以用 SET alert:recover:lock 加过期时间加分布式锁；更严谨时，可以用 Lua 脚本把“判断超时、删除索引、重新入队”做成原子操作。

避坑总结：至少要避开这些线上问题 ​

1. 只 BRPOPLPUSH 不 LREM，处理中队列无限增长 ​

常见现象是服务运行几天后 LLEN alert:processing:worker-1 持续变大，内存不断上涨。原因通常是业务成功后没有执行确认删除，或者异常分支直接 return，跳过了 LREM。修正方法是把确认逻辑放到 finally 或统一出口，并对 processing 队列长度建立监控。

2. 队列里直接存大 JSON，失败重试只能模糊匹配 ​

有些实现把完整 JSON 从 pending 移到 processing，失败后用 LREM key 1 json 删除。只要 JSON 字段顺序变化、空格不同或内容重复，就可能删错或删不到。正确做法是 List 中放唯一 ID，完整内容放 alert:meta:{id}，删除时始终按 ID 精确匹配。

3. 失败后立刻无限重入队，造成消息热循环 ​

错误现象是 Redis QPS 飙高、通知服务被同一条坏告警反复打爆。原因可能是下游参数非法、通知模板缺字段，这类错误重试再多次也不会成功。应区分临时错误和永久错误，使用 HINCRBY 限制重试次数，超过阈值进入 alert:dead，并保留元数据供排查。

4. 多个消费者共用一个 processing 队列，回收责任不清 ​

多个 worker 共用 alert:processing 时，某个 worker 宕机后，其他实例很难判断消息归属，也可能误删别人的处理中消息。建议按消费者命名 alert:processing:{worker_id}，回收器结合 worker 心跳判断存活；已经下线的 worker，其队列才能被统一接管。

5. 把 BRPOPLPUSH 当成严格 Exactly-Once ​

BRPOPLPUSH 保证移动动作原子，但不能覆盖业务通知的重复发送。例如消费者已经发出通知，但在执行 LREM 前被 kill，回收器会再次投递，最终可能重复告警。通知端应按告警指纹和时间窗口做幂等，关键工单系统携带事件唯一 ID。

结语 ​

Redis List 并不是过时的数据结构。在告警收敛这种“轻量队列 + 短时聚合 + 必须可回溯”的场景里，List 加上 BRPOPLPUSH 可以用很低成本搭建可靠管线：pending 负责缓冲，processing 负责安全备份，dead 负责失败回溯，meta 负责上下文还原。等业务发展到多消费组、长期保留和范围回放时，再迁移到 Stream 也不迟。

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。

本文来自开发者日记（devdiary.cn）· 记录开发者实战笔记与免费在线工具箱

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。