Redis Stream消息队列消费组、Pending列表与消息可靠性的工程实践「我们用 Redis 做消息队列」——这句话在三年前意味着「用 List 的 BLPOP 拉消息丢了你别找我」。Redis 5.0 引入 Stream 之后这句话的含义彻底变了。一、Stream 的数据结构不止是「追加写」日志Redis Stream 在底层是一个**基数树Radix Tree**组织的宏节点Macro Node链表而非简单的 linked list。每个 Stream 包含多个宏节点每个宏节点内存储多条消息消息按 ID 严格递增。消息 ID 的格式为millisecondsTime-sequenceNumber例如1718600000123-0。这个自增 ID 的设计赋予了 Stream 两个核心能力范围查询XRANGE mystream 1718500000000 1718600000000可以按时间范围拉取消息不需要从队列头部开始顺序遍历。回溯消费不同于 Kafka 按 offset 消费Stream 的消费者可以从任意 ID 开始读取实现真正的「时间旅行」式消费。flowchart TB subgraph StreamStructure[Redis Stream 内部结构] direction TB RadixTree[基数树索引\n(按消息 ID 索引)] subgraph MacroNode1[宏节点 #1] M1[消息 1718600000000-0\n{order_id: 1001, action: create}] M2[消息 1718600000001-0\n{order_id: 1002, action: pay}] M3[消息 1718600000002-0\n{order_id: 1003, action: ship}] end subgraph MacroNode2[宏节点 #2] M4[消息 1718600000003-0\n{order_id: 1004, action: confirm}] M5[消息 1718600000004-0\n{order_id: 1005, action: refund}] end end RadixTree -- MacroNode1 RadixTree -- MacroNode2 subgraph ConsumerGroup[消费组: order-processors] CG[last_delivered_id: 1718600000001-0] PEL[Pending Entries List:\n- 1718600000002-0 → consumer-A (idle: 32s)\n- 1718600000003-0 → consumer-B (idle: 8s)] end CG -- M3 CG -- M4二、消费组与游标管理XREADGROUP 的精确控制消费组Consumer Group是 Stream 区别于传统 Redis List 的核心特性。它解决了三个问题消息不会因为消费者挂掉而丢失——Pending Entries List 兜底多消费者负载均衡——同组内消息不会重复投递消费进度的持久化——游标last_delivered_id存储在 Redis 服务端以下是生产级的消费组配置与消费循环package com.example.stream; import io.lettuce.core.*; import io.lettuce.core.api.StatefulRedisConnection; import io.lettuce.core.api.sync.RedisCommands; import java.time.Duration; import java.util.List; import java.util.Map; /** * Redis Stream 消费组 —— 生产级消费者实现。 * 核心特性自动 ACK、Pending 消息认领、死信队列。 */ public class StreamConsumerGroup { private static final String STREAM_KEY orders:events; private static final String GROUP_NAME order-processors; private static final String CONSUMER_NAME instance- java.net.InetAddress.getLocalHost().getHostName(); private static final String DEAD_LETTER_STREAM STREAM_KEY :dead; // 消费参数 private static final int BATCH_SIZE 10; // 每批消费 10 条 private static final Duration BLOCK_TIMEOUT Duration.ofSeconds(2); private static final Duration PENDING_CLAIM_IDLE Duration.ofSeconds(60); private final RedisCommandsString, String commands; public StreamConsumerGroup(RedisClient redisClient) { StatefulRedisConnectionString, String connection redisClient.connect(); this.commands connection.sync(); initialize(); } /** * 初始化创建消费组幂等操作。 * MKSTREAM 确保 Stream 不存在时自动创建。 * 消费组从 Stream 头部开始消费0-0避免漏掉历史消息。 */ private void initialize() { try { commands.xgroupCreate( XReadArgs.StreamOffset.from(STREAM_KEY, 0-0), GROUP_NAME, XGroupCreateArgs.Builder.mkstream(true) // 不存在则创建 Stream ); } catch (RedisCommandExecutionException e) { // BUSYGROUP 错误表示消费组已存在忽略即可 if (!e.getMessage().contains(BUSYGROUP)) { throw e; } } } /** * 消费主循环。生产环境中在独立线程中运行。 */ public void consumeLoop() { while (!Thread.currentThread().isInterrupted()) { try { // 1. 先认领 Pending 超过 60 秒的消息可能属于已宕机的消费者 claimPendingMessages(); // 2. 从消费组拉取新消息 ListStreamMessageString, String messages commands.xreadgroup( Consumer.from(GROUP_NAME, CONSUMER_NAME), XReadArgs.Builder.block(BLOCK_TIMEOUT).count(BATCH_SIZE), XReadArgs.StreamOffset.lastConsumed(STREAM_KEY) ); if (messages null || messages.isEmpty()) { continue; // 无新消息继续等待 } for (StreamMessageString, String msg : messages) { processMessage(msg); } } catch (RedisException e) { // Redis 连接异常等待重连不要疯狂重试 // logger.error(Redis connection lost, will retry after backoff, e); sleepSafely(5000); } catch (Exception e) { // 其他未知异常记录并继续 // logger.error(Unexpected error in consume loop, e); } } } /** * 认领 Pending 超过阈值但未被 ACK 的消息。 * 场景消费者 A 拿到了消息但进程崩溃消息卡在 PEL 中。 * 通过 XAUTOCLAIM 将这些消息转移给当前消费者重新处理。 */ private void claimPendingMessages() { try { // XAUTOCLAIM自动扫描 PEL将闲置超过 60 秒的消息转移给当前消费者 commands.xautoclaim( STREAM_KEY, XAutoClaimArgs.Builder .consumer(Consumer.from(GROUP_NAME, CONSUMER_NAME)) .minIdleTime(PENDING_CLAIM_IDLE) .count(100) ); } catch (Exception e) { // 无 Pending 消息或 Stream 为空时忽略错误 // logger.debug(No pending messages to claim, e); } } /** * 处理单条消息带异常处理和死信机制。 */ private void processMessage(StreamMessageString, String msg) { String messageId msg.getId(); MapString, String body msg.getBody(); try { // 业务处理逻辑 handleOrderEvent(body); // 处理成功确认 ACK从 PEL 中移除 commands.xack(STREAM_KEY, GROUP_NAME, messageId); } catch (NonRetryableException e) { // 不可重试异常如数据格式错误直接 ACK 进入死信队列 // logger.error(Non-retryable error for message {}, messageId, e); commands.xack(STREAM_KEY, GROUP_NAME, messageId); sendToDeadLetter(messageId, body, e.getMessage()); } catch (RetryableException e) { // 可重试异常如下游服务临时不可用不 ACK让消息留在 PEL // 下一次 XAUTOCLAIM 会自动认领重试 // logger.warn(Retryable error for message {}, keeping in PEL, messageId, e); } } /** * 死信队列将无法处理的消息写入独立的 Stream供后续人工排查。 */ private void sendToDeadLetter(String originalId, MapString, String body, String error) { try { MapString, String deadBody new java.util.HashMap(body); deadBody.put(_original_id, originalId); deadBody.put(_error, error); deadBody.put(_timestamp, String.valueOf(System.currentTimeMillis())); commands.xadd(DEAD_LETTER_STREAM, deadBody); } catch (Exception e) { // 死信写入失败是最坏情况记录到本地日志作为最后兜底 // logger.error(CRITICAL: Failed to write to dead letter stream. msg{}, originalId, e); } } private void handleOrderEvent(MapString, String body) throws RetryableException, NonRetryableException { // 实际业务逻辑 } private void sleepSafely(long millis) { try { Thread.sleep(millis); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } // --- 自定义异常 --- private static class RetryableException extends Exception {} private static class NonRetryableException extends Exception {} }三、消息确认机制的深度解析Redis Stream 的确认机制与 Kafka 的 Offset 提交有本质区别特性Redis StreamKafka确认粒度单条消息Offset批次未确认消息存储Pending Entries List服务端维护Consumer 本地 Offset 管理消费者宕机后XAUTOCLAIM 自动认领Rebalance 从 Committed Offset 恢复消息回溯能力任意 ID 范围查询从 Committed Offset 开始Stream 的 Ack 是精确到单条消息的这意味着你可以选择性确认batch 中只 ACK 成功的 9 条失败的那条留在 PEL不会有 Kafka 中「一个 offset 提交成功整批消息被标记为已消费」的粗粒度问题但代价是Stream 需要维护 PEL 的额外内存开销。每个未被 ACK 的消息会在 PEL 中保存一个条目包含消息 ID、消费者名称和闲置时间。如果大量消费者宕机且长时间不恢复PEL 会持续增长影响性能。四、Stream vs Kafka功能边界与选型建议一个高频问题「既然 Stream 支持消费组和持久化能不能替代 Kafka」答案是不能也不应该。它们在设计哲学上有根本差异graph LR subgraph Redis[Redis Stream 适用场景] R1[轻量级异步任务 10万 msg/s] R2[已有的 Redis 基础设施] R3[需要消息回溯 简单可靠性的场景] R4[小团队不想运维 Kafka] end subgraph Kafka[Kafka 适用场景] K1[海量事件流 100万 msg/s] K2[需要长时间消息持久化天/周级] K3[多消费者组独立消费同一 Topic] K4[需要 Exactly-Once 语义 事务] end style R1 fill:#cfc,stroke:#333 style R2 fill:#cfc,stroke:#333 style R3 fill:#cfc,stroke:#333 style R4 fill:#cfc,stroke:#333 style K1 fill:#f96,stroke:#333 style K2 fill:#f96,stroke:#333 style K3 fill:#f96,stroke:#333 style K4 fill:#f96,stroke:#333Stream 的最佳实践场景异步任务分发如订单状态变更通知、邮件发送队列消息量在每秒万级以内对持久化要求是「小时级」而非「天级」。如果把这些场景用 Kafka 来做运维成本远高于收益。反过来如果你的场景是用户行为埋点流日均千亿条消息需要多团队独立消费——请用 Kafka。Stream 不是为这个量级设计的。五、总结Redis Stream 补齐了 Redis 作为消息队列的最后一环消费组解决了消息不丢失PEL 机制让消费者宕机不再是灾难XAUTOCLAIM 自动认领闲置消息。ACK 粒度是单条而非批次选择性确认赋予业务更精细的控制力。死信队列是必须的不可重试的消息不应当永远卡在 PEL 中消费系统的资源独立写入死信 Stream 供人工排查。Stream 不是 Kafka 的替代品选型要基于消息量级、持久化需求和运维成本三个维度综合决策。如果你的系统已经在用 Redis而异步任务量在可控范围内Stream 是比「再搭一套 Kafka 集群」务实得多的选择。