HOOOS

如何基于 Redis Stream 构建高可靠死信队列（DLQ）机制

0 38 架构师老王 RedisRedis Stream死信队列消息队列分布式系统
Apple

在构建基于消息队列的分布式系统时，处理失败的消息是一个绕不开的问题。反复失败的消息如果不能被妥善处理，可能会阻塞正常消息的处理流程，甚至耗尽系统资源。死信队列（Dead Letter Queue, DLQ）是一种常见的解决方案，用于隔离和处理这些“顽固”的失败消息。本文将详细探讨如何利用 Redis Stream 的特性，设计并实现一个可靠、高效的 DLQ 机制。

为什么选择 Redis Stream？

Redis Stream 是 Redis 5.0 引入的强大的数据结构，它借鉴了 Kafka 等专业消息队列的设计，提供了持久化、消费者组、消息确认（ACK）等关键特性，非常适合作为消息队列使用。相比于 List 或 Pub/Sub，Stream 提供了更完善的消息处理语义，尤其是消费者组和 XPENDING / XCLAIM 机制，为实现 DLQ 提供了基础。

DLQ 核心流程设计

我们的目标是：当一个消息在主 Stream 中被消费者组（Consumer Group）处理失败，并且重试次数达到预设阈值后，能自动、原子地将其转移到专门的 DLQ Stream 中，以便后续进行分析、告警或手动干预。

核心流程可以概括为：

  1. 消费消息：消费者通过 XREADGROUP 从主 Stream 读取消息。
  2. 处理消息：消费者尝试处理消息。
  3. 成功确认：处理成功，消费者使用 XACK 确认消息。
  4. 失败处理：处理失败，消费者不 XACK。消息会保留在 Pending Entries List (PEL) 中。
  5. 超时认领：通过一个独立的任务（或集成在消费者逻辑中），定期检查 PEL 中超时的消息（使用 XPENDING）。
  6. 重新投递/认领：使用 XCLAIM 尝试重新将超时的消息分配给当前或其他消费者进行重试。
  7. 判断是否进入 DLQ：在 XCLAIM 之后（或者在消费者处理 XCLAIM 获得的消息时），检查消息的投递次数（Delivery Count）。
  8. 原子转移：如果投递次数超过阈值，则调用一个 Lua 脚本，原子地将消息添加到 DLQ Stream 并从主 Stream 中 XACK（或 XDEL，取决于业务语义）。
  9. DLQ 监控与处理：独立的监控程序或消费者组负责处理 DLQ Stream 中的消息。
graph LR
    A[Consumer: XREADGROUP] --> B{Process Message};
    B -- Success --> C[Consumer: XACK];
    B -- Failure --> D[Message stays in PEL];
    E[Monitor/Claimer: XPENDING] --> F{Find Idle Messages};
    F -- Found --> G[Monitor/Claimer: XCLAIM];
    G --> H[Consumer: Receives Claimed Message];
    H --> I{Check Delivery Count};
    I -- Count < Threshold --> B;
    I -- Count >= Threshold --> J[Atomic Transfer to DLQ (Lua Script)];
    J --> K[Message in DLQ Stream];
    J --> L[XACK Original Message];
    M[DLQ Monitor/Processor] --> K;

(上述为 Mermaid 流程图描述，实际渲染需相应工具)

关键实现细节

1. 判断消息进入 DLQ 的时机与依据

这是 DLQ 机制的核心。我们主要依赖 Redis Stream 提供的 XPENDING 和 XCLAIM 命令。

  • XPENDING stream group [idle min-idle-time] [start end count] [consumer]
    这个命令可以查询指定消费者组中处于 Pending 状态（已被读取但未 ACK）的消息列表。关键信息包括：消息 ID、所属消费者、空闲时间（idle time） 和 投递次数（delivery count）。

  • XCLAIM stream group consumer min-idle-time message-id [message-id ...]
    这个命令用于改变 PEL 中消息的所有权，通常用于处理那些长时间未被 ACK 的“僵尸”消息。当一个消息被 XCLAIM 后，它的投递次数（delivery count）会自动加 1。

判断逻辑：

我们可以在以下两个时机检查投递次数：

  • 认领时检查 (Proactive Check)：一个专门的监控/认领服务定期执行 XPENDING 获取超时的消息列表。对每个超时的消息，它会检查其当前的 delivery_count。如果 delivery_count 已经达到或超过阈值（比如 3 次），则不执行 XCLAIM，而是直接触发转移到 DLQ 的逻辑。
  • 消费时检查 (Reactive Check)：消费者在处理通过 XREADGROUP 或 XCLAIM 获取的消息时，检查该消息的投递次数。这需要一种方法让消费者知道当前消息的投递次数。XREADGROUP 返回的消息本身不直接包含投递次数，但 XCLAIM 后再次读取或通过 XPENDING 查询可以获得。更常见的是，认领者（Claimer）执行 XCLAIM，然后被 XCLAIM 的消息会被重新投递给某个消费者（可能是同一个或不同的）。这个消费者在处理这条消息时，需要再次查询其投递次数（例如，通过 XPENDING stream group - + 1 message-id）。如果次数超限，则触发 DLQ 转移。

推荐方式：认领时检查 (Proactive Check)

这种方式更清晰，将 DLQ 判断逻辑集中在认领/监控服务中。流程如下：

  1. 监控服务定期调用 XPENDING 获取指定 idle 时间以上且 delivery_count 小于阈值的消息列表。
  2. 对这些消息执行 XCLAIM，将它们重新分配给活跃的消费者。
  3. 监控服务再次调用 XPENDING 获取指定 idle 时间以上且 delivery_count 大于等于阈值的消息列表。
  4. 对这些达到阈值的消息，执行原子转移到 DLQ 的操作。

示例 XPENDING 查询达到阈值的消息：

假设主 Stream 名为 mystream，消费者组为 mygroup，DLQ 阈值为 3 次。

# 查询空闲超过 60 秒的消息
XPENDING mystream mygroup IDLE 60000 - + 10

在返回的结果中，每一条消息都会包含其投递次数。监控服务需要解析这个结果，筛选出 delivery_count >= 3 的消息 ID。

1) 1) "1678886400000-0"
   2) "consumer-a"
   3) (integer) 75000  # Idle time in ms
   4) (integer) 3      # Delivery count <---- 达到阈值!
2) 1) "1678886401000-0"
   2) "consumer-b"
   3) (integer) 65000
   4) (integer) 1

2. 原子性转移消息到 DLQ

将消息从主 Stream 转移到 DLQ Stream 必须是原子操作，以防止消息丢失或重复进入 DLQ。这包括两个步骤：

  1. 将原始消息内容添加到 DLQ Stream。
  2. 在主 Stream 中确认（XACK）原始消息，将其从 PEL 中移除。

这两个操作必须捆绑在一起执行，要么都成功，要么都失败。Lua 脚本 是实现这种原子性的最佳方式。

Lua 脚本逻辑 (transfer_to_dlq.lua)

-- KEYS[1]: 主 Stream 名称 (e.g., mystream)
-- KEYS[2]: 消费者组名称 (e.g., mygroup)
-- KEYS[3]: DLQ Stream 名称 (e.g., mystream:dlq)
-- ARGV[1]: 需要转移的消息 ID

local mainStream = KEYS[1]
local groupName = KEYS[2]
local dlqStream = KEYS[3]
local messageId = ARGV[1]

-- 1. 尝试读取原始消息内容
--    注意: XREAD 不保证一定能读到，特别是如果消息已被 XDEL。但通常 PEL 里的消息还在。
--    更稳妥的方式是，在调用脚本前，监控服务已经通过 XPENDING 或 XCLAIM 获取了消息内容。
--    这里我们假设调用者会传入消息内容，或者我们直接用 XREVRANGE 读取。
local messageData = redis.call('XREVRANGE', mainStream, messageId, messageId, 'COUNT', 1)

if not messageData or #messageData == 0 then
  -- 如果消息已被删除（可能被 XDEL），则直接尝试 ACK，避免脚本错误
  -- 这种情况理论上不常见，因为我们是处理 PEL 中的消息
  local ackResult = redis.call('XACK', mainStream, groupName, messageId)
  return {ok='Message not found, ACK attempted anyway', ack_result=ackResult}
end

-- 提取消息的 fields and values
-- messageData 格式: { {messageId, {field1, value1, field2, value2, ...}} }
local fieldsAndValues = messageData[1][2]

-- 2. 将消息添加到 DLQ Stream
--    为了保留原始 ID 信息，可以将其作为一个字段添加到 DLQ 消息中
--    也可以添加其他元数据，如失败时间、原始 Stream 等
local dlqPayload = {}
table.insert(dlqPayload, 'original_message_id')
table.insert(dlqPayload, messageId)
table.insert(dlqPayload, 'transfer_time')
table.insert(dlqPayload, redis.call('TIME')[1]) -- 秒级时间戳
-- 将原始消息的字段和值追加到 DLQ payload
for i=1, #fieldsAndValues do
    table.insert(dlqPayload, fieldsAndValues[i])
end

local addResult = redis.call('XADD', dlqStream, '*', unpack(dlqPayload))

if not addResult then
  return {err='Failed to add message to DLQ stream ' .. dlqStream}
end

-- 3. 在主 Stream 中确认原始消息
local ackResult = redis.call('XACK', mainStream, groupName, messageId)

if ackResult == 1 then
  return {ok='Message transferred to DLQ and ACKed', dlq_message_id=addResult, acked_count=ackResult}
else
  -- ACK 失败通常意味着消息 ID 不在 PEL 中，可能是并发操作导致，或者已被 ACK
  -- 只要 DLQ 添加成功，也算部分成功，需要记录日志
  return {warn='Message added to DLQ, but failed to ACK in main stream (already ACKed or not in PEL?)', dlq_message_id=addResult, acked_count=ackResult}
end

如何调用 Lua 脚本：

首先，加载脚本到 Redis：

SCRIPT LOAD "$(cat transfer_to_dlq.lua)"
# 返回脚本的 SHA1 哈希值，例如：abcdef1234567890abcdef1234567890abcdef12

然后，监控服务在确定某个消息需要进入 DLQ 时，使用 EVALSHA 调用：

EVALSHA abcdef1234567890abcdef1234567890abcdef12 3 mystream mygroup mystream:dlq 1678886400000-0

思考点：

  • 消息内容获取：上面的脚本尝试用 XREVRANGE 读取消息。如果担心消息可能在主 Stream 中被 XTRIM 或 XDEL 清理掉，监控服务在决定转移前，就应该先读取并缓存消息内容，然后将内容作为参数传递给 Lua 脚本。这样脚本就不需要读取主 Stream，只需要 XADD 到 DLQ 和 XACK 主 Stream。
  • 错误处理：Lua 脚本中的错误处理需要仔细考虑。如果 XADD 成功但 XACK 失败，消息会留在 DLQ 中，但也可能被再次 XCLAIM。需要确保 DLQ 的处理具有幂等性，或者有机制识别这种情况。
  • 性能：频繁执行 Lua 脚本会对 Redis 性能产生影响。监控服务的执行频率、每次处理的消息量需要根据实际负载调整。

3. 监控和处理 DLQ

DLQ 中的消息通常表示需要人工介入或特殊逻辑处理的异常情况。如何处理 DLQ 取决于具体的业务需求。

  • 独立消费者组：可以为 DLQ Stream 创建一个独立的消费者组 (dlq_group) 和消费者 (dlq_processor)。这个消费者负责读取 DLQ 消息，进行分析、记录日志、发送告警，或者尝试修复数据后重新投递到主 Stream（需要非常小心，避免死循环）。

    # 创建 DLQ 消费者组 (如果不存在)
XGROUP CREATE mystream:dlq dlq_group 0 MKSTREAM

# DLQ 处理器读取消息
XREADGROUP GROUP dlq_group dlq_processor COUNT 10 BLOCK 5000 STREAMS mystream:dlq >

  • 定期轮询/导出：如果 DLQ 的消息量不大，或者不需要实时处理，也可以编写一个脚本定期使用 XRANGE 或 XREAD 读取 DLQ 中的所有消息，然后进行批量处理、导出到日志系统或数据库进行分析。

    # 读取 DLQ Stream 中所有消息
XRANGE mystream:dlq - +

  • 告警：集成监控系统（如 Prometheus + Alertmanager），监控 DLQ Stream 的长度 (XLEN mystream:dlq)。当长度超过阈值或持续增长时，触发告警通知运维或开发人员。

  • 自动重试/修复？：某些情况下，可以尝试编写自动修复逻辑。例如，如果失败原因是暂时的外部服务不可用，DLQ 处理器可以等待一段时间后，将消息重新格式化并投递回主 Stream 的入口。但这需要非常谨慎的设计，确保不会导致无限循环。

DLQ 消息处理要点：

  • 幂等性：DLQ 处理逻辑应设计为幂等的，即使同一条死信消息被重复处理，结果也应一致。
  • 信息丰富：存入 DLQ 时，尽量包含足够多的上下文信息，如原始消息 ID、失败原因（如果消费者能提供）、失败时间戳、重试次数、原始 Stream 等，方便排查问题。
  • 清理机制：DLQ Stream 也需要管理大小，防止无限增长。可以根据消息时间戳或 Stream 长度设置 XTRIM 策略，定期清理已被处理或过于陈旧的死信消息。

实践中的考量

  • 监控服务的健壮性：负责检查 PEL、执行 XCLAIM 和调用 Lua 脚本的监控服务本身需要是高可用的。可以部署多个实例，通过分布式锁（如 Redlock）确保只有一个实例在工作。
  • 阈值设置：重试次数阈值需要根据业务容忍度和失败类型来设定。太小可能导致过多消息进入 DLQ，太大则可能延迟发现问题。
  • 性能：XPENDING 操作在 PEL 很大时可能会变慢。Redis 6.2 对 XPENDING 进行了优化，但仍需关注其性能影响。考虑是否需要分片或更精细的查询范围。
  • Redis 版本：确保使用的 Redis 版本支持 Stream 相关的所有命令（最低 5.0，推荐 6.0+ 以获得更好的性能和特性）。
  • Lua 脚本超时：复杂的 Lua 脚本或在 Redis 繁忙时执行可能超过默认的 lua-time-limit。需要监控脚本执行时间，必要时进行优化或调整配置。

总结

利用 Redis Stream 的消费者组、XPENDING、XCLAIM 和 Lua 脚本，我们可以构建一个功能完善且可靠的死信队列机制。核心在于准确追踪消息的投递次数，并通过 Lua 脚本实现原子性的消息转移。合理的 DLQ 设计能够有效隔离处理失败的消息，提高系统的健壮性和可维护性，是构建复杂分布式消息系统的关键一环。记住，DLQ 不是银弹，后续对 DLQ 中消息的有效监控和处理同样重要。

点评评价

captcha
健康