Redis Streams 与消费者组：可靠的事件处理

1. 概览

Redis 的 Pub/Sub 有一个大缺点。如果消费者在发布时处于离线状态，消息就会丢失。Redis Streams 解决了这个问题。消息被持久化存储在流中，每个消费者组都会记住哪些消息已经被处理。

本文指导你使用 Spring Boot 3 和 Spring Data Redis 构建订单事件处理流程。流程包括几个步骤：写入事件、通过消费者组读取、手动 ACK、重新处理待处理消息，以及将失败消息转移到死信流。

2. Redis Streams 和消费者组如何工作

在写 Java 之前，你应该在 redis-cli 中尝试这些原始命令：

XADD orders * orderId 1001 amount 250000
XGROUP CREATE orders shipping-group 0
XREADGROUP GROUP shipping-group worker-1 COUNT 1 STREAMS orders >
XPENDING orders shipping-group
XACK orders shipping-group 1728640000000-0

各个命令的含义：

• XADD 添加一个条目到 orders 流。* 符号让 Redis 自动生成 <millis>-<seq> 形式的 ID。
• XGROUP CREATE ... 0 创建 shipping-group 消费者组，并从流的开头开始读取。如果使用 $，消费者组只会读取新消息。
• XREADGROUP ... > 获取组内尚未分配给任何人的消息。消息被分配给 worker-1，同时被记录在该组的待处理条目列表 (PEL) 中。
• XPENDING 显示已分配但尚未 ACK 的消息数量。
• XACK 从 PEL 中删除消息。从此刻起，消费者组认为该消息已被处理。

XPENDING 在 ACK 前的输出如下所示：

1) (integer) 1
2) "1728640000000-0"
3) "1728640000000-0"
4) 1) 1) "worker-1"
      2) "1"

PEL 是可靠性的关键。假设消费者在收到消息后但在 ACK 前崩溃了。消息仍然留在 PEL 中，另一个消费者可以认领 (claim) 它来继续处理。这就是 at-least-once 语义的体现。

3. 安装

添加 Spring Data Redis 依赖。此 starter 默认使用 Lettuce 作为客户端：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

配置连接和流名称：

spring:
  data:
    redis:
      host: localhost
      port: 6379

app:
  stream:
    key: orders
    group: shipping-group
    consumer: ${HOSTNAME:worker-local}

消费者名称来自 HOSTNAME，所以每个实例（pod）都有唯一的名称。这很重要。如果两个实例使用相同的名称，Redis 会将它们视为一个消费者，它们的 PEL 会相互混淆。

4. 将事件写入流

@Service
public class OrderEventPublisher {

    private final StringRedisTemplate redisTemplate;
    private final String streamKey;

    public OrderEventPublisher(StringRedisTemplate redisTemplate,
                               @Value("${app.stream.key}") String streamKey) {
        this.redisTemplate = redisTemplate;
        this.streamKey = streamKey;
    }

    public RecordId publishOrderCreated(String orderId, long amount) {
        MapRecord<String, String, String> record = StreamRecords.newRecord()
                .in(streamKey)
                .ofMap(Map.of(
                        "type", "ORDER_CREATED",
                        "orderId", orderId,
                        "amount", String.valueOf(amount)));

        return redisTemplate.opsForStream().add(record);
    }
}

StreamRecords.newRecord().in(...).ofMap(...) 创建一个字段为字符串的 MapRecord。opsForStream().add() 命令等同于 XADD orders * ...，并返回 Redis 生成的 RecordId，例如 1728640000000-0。

使用 StringRedisTemplate 确保 Redis 中的数据可以用 redis-cli 读取。你不需要处理 JDK 序列化的字节串。

5. 安全地创建消费者组

消费者组必须存在才能读取。应用启动时流可能还没有任何消息，所以你需要 MKSTREAM 参数。StreamOperations 的 createGroup() 方法不启用此参数。因此我们需要调用更底层的连接：

@Component
public class StreamGroupInitializer {

    private final StringRedisTemplate redisTemplate;

    public StreamGroupInitializer(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    public void ensureGroup(String streamKey, String group) {
        try {
            redisTemplate.execute((RedisCallback<String>) connection ->
                    connection.streamCommands().xGroupCreate(
                            streamKey.getBytes(StandardCharsets.UTF_8),
                            group,
                            ReadOffset.from("0"),
                            true)); // MKSTREAM
        } catch (DataAccessException e) {
            String msg = NestedExceptionUtils.getMostSpecificCause(e).getMessage();
            if (msg == null || !msg.contains("BUSYGROUP")) {
                throw e;
            }
            // 消费者组已存在：应用重启时正常
        }
    }
}

最后一个参数 true 启用 MKSTREAM。如果流不存在，Redis 将创建一个空流。应用重启时，Redis 返回 BUSYGROUP Consumer Group name already exists 错误。我们忽略这个特定的错误，但重新抛出其他所有错误。

6. 使用 StreamMessageListenerContainer 读取消息

6.1. 支持手动 ACK 的 Listener

@Component
public class OrderEventListener
        implements StreamListener<String, MapRecord<String, String, String>> {

    private final StringRedisTemplate redisTemplate;
    private final ShippingService shippingService;
    private final String streamKey;
    private final String group;

    public OrderEventListener(StringRedisTemplate redisTemplate,
                              ShippingService shippingService,
                              @Value("${app.stream.key}") String streamKey,
                              @Value("${app.stream.group}") String group) {
        this.redisTemplate = redisTemplate;
        this.shippingService = shippingService;
        this.streamKey = streamKey;
        this.group = group;
    }

    @Override
    public void onMessage(MapRecord<String, String, String> record) {
        Map<String, String> body = record.getValue();
        shippingService.createShipment(body.get("orderId"));

        redisTemplate.opsForStream().acknowledge(streamKey, group, record.getId());
    }
}

ACK 命令只在 createShipment 成功时运行。如果该方法抛出异常，消息仍然留在 PEL 中，以便稍后重新处理。不要在处理前 ACK。这样做会将 at-least-once 变成 at-most-once。

本文中的 ShippingService 是一个简单的服务。它记录已创建发货单的 orderId：

@Service
public class ShippingService {

    private final Set<String> shippedOrders = ConcurrentHashMap.newKeySet();

    public void createShipment(String orderId) {
        if (!shippedOrders.add(orderId)) {
            return; // 已经处理过了，跳过（幂等性）
        }
        // 调用快递公司 API...
    }

    public Set<String> shippedOrders() {
        return Set.copyOf(shippedOrders);
    }
}

6.2. 容器配置

@Configuration
public class StreamConfig {

    @Bean(destroyMethod = "stop")
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>>
    orderStreamContainer(RedisConnectionFactory connectionFactory,
                         StreamGroupInitializer initializer,
                         OrderEventListener listener,
                         @Value("${app.stream.key}") String streamKey,
                         @Value("${app.stream.group}") String group,
                         @Value("${app.stream.consumer}") String consumerName) {

        initializer.ensureGroup(streamKey, group);

        var options = StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                .pollTimeout(Duration.ofSeconds(2))
                .batchSize(10)
                .build();

        var container = StreamMessageListenerContainer.create(connectionFactory, options);

        var request = StreamMessageListenerContainer.StreamReadRequest
                .builder(StreamOffset.create(streamKey, ReadOffset.lastConsumed()))
                .consumer(Consumer.from(group, consumerName))
                .autoAcknowledge(false)
                .cancelOnError(t -> false)
                .build();

        container.register(request, listener);
        container.start();
        return container;
    }
}

配置中的关键点：

• ReadOffset.lastConsumed() 对应 XREADGROUP 中的 >，即只获取该消费者组的新消息。
• pollTimeout 是每次读取的阻塞时间。batchSize 是一次读取的最大消息数。
• autoAcknowledge(false) 禁用自动 ACK，因为我们在 listener 中手动 ACK。
• cancelOnError(t -> false) 是最重要的一行。默认情况下，当 listener 抛出异常时，容器取消订阅。消费者会悄悄地停止读取，没有任何明确的告警。

注意： 对于重要的业务逻辑，不要使用 receiveAutoAck(...)。这种模式会在读取时立即 ACK，也就是在 listener 执行前。如果应用在中间崩溃，消息将永久丢失。

7. 处理待处理消息和死信

Listener 只读取新消息（>）。一条失败的消息或来自已死亡消费者的消息将永远留在 PEL 中，除非有人处理它。你需要一个定期执行的任务来处理这种情况：

@Component
public class PendingOrderReclaimer {

    private static final Duration MIN_IDLE = Duration.ofMinutes(1);
    private static final long MAX_DELIVERIES = 5;

    private final StringRedisTemplate redisTemplate;
    private final ShippingService shippingService;
    private final String streamKey;
    private final String group;
    private final String consumerName;

    public PendingOrderReclaimer(StringRedisTemplate redisTemplate,
                                 ShippingService shippingService,
                                 @Value("${app.stream.key}") String streamKey,
                                 @Value("${app.stream.group}") String group,
                                 @Value("${app.stream.consumer}") String consumerName) {
        this.redisTemplate = redisTemplate;
        this.shippingService = shippingService;
        this.streamKey = streamKey;
        this.group = group;
        this.consumerName = consumerName;
    }

    @Scheduled(fixedDelay = 30_000)
    public void reclaim() {
        StreamOperations<String, String, String> ops = redisTemplate.opsForStream();
        PendingMessages pending = ops.pending(streamKey, group, Range.unbounded(), 50);

        for (PendingMessage pm : pending) {
            if (pm.getElapsedTimeSinceLastDelivery().compareTo(MIN_IDLE) < 0) {
                continue;
            }
            List<MapRecord<String, String, String>> claimed =
                    ops.claim(streamKey, group, consumerName, MIN_IDLE, pm.getId());

            for (MapRecord<String, String, String> record : claimed) {
                if (pm.getTotalDeliveryCount() >= MAX_DELIVERIES) {
                    moveToDeadLetter(record);
                } else {
                    shippingService.createShipment(record.getValue().get("orderId"));
                }
                ops.acknowledge(streamKey, group, record.getId());
            }
        }
    }

    private void moveToDeadLetter(MapRecord<String, String, String> record) {
        Map<String, String> body = new HashMap<>(record.getValue());
        body.put("originalId", record.getId().getValue());
        redisTemplate.opsForStream().add(
                StreamRecords.newRecord().in(streamKey + ":dlq").ofMap(body));
    }
}

任务的处理流程：

1. pending(...) 等同于 XPENDING orders shipping-group - + 50。每个 PendingMessage 告诉你哪个消费者正在处理该消息、空闲时间和已分配次数。
2. claim(...) 等同于 XCLAIM。该命令将所有权转移给当前消费者。Redis 只在消息空闲至少 MIN_IDLE 时才会转移，因此两个实例同时运行该任务时不会认领重复的消息。每次认领也会增加投递计数。
3. 如果消息被分配超过 MAX_DELIVERIES 次，它被复制到 orders:dlq 流，然后才 ACK。这样，永久有问题的消息（poison message）就不会被无限重试。

如果 createShipment 抛出异常，ACK 命令不会运行。消息留在 PEL 中，在下一个周期重试。记得在 @Configuration 类上启用 @EnableScheduling。

注意： 从 Redis 6.2 开始，有 XAUTOCLAIM 命令。该命令将 XPENDING 和 XCLAIM 步骤合并为一个，当 PEL 很大时效率更高。如果你使用的 Spring Data Redis 版本还没有便利的 API，你可以通过 redisTemplate.execute(...) 调用它。

8. 使用 Testcontainers 编写测试

为测试添加依赖：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-testcontainers</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>junit-jupiter</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.awaitility</groupId>
    <artifactId>awaitility</artifactId>
    <scope>test</scope>
</dependency>

下面的测试发布一个事件。然后它检查发货单已被创建且 PEL 为空：

@SpringBootTest
@Testcontainers
class OrderStreamIntegrationTest {

    @Container
    @ServiceConnection(name = "redis")
    static GenericContainer<?> redis =
            new GenericContainer<>("redis:7-alpine").withExposedPorts(6379);

    @Autowired OrderEventPublisher publisher;
    @Autowired ShippingService shippingService;
    @Autowired StringRedisTemplate redisTemplate;

    @Test
    void whenOrderPublished_thenShipmentCreatedAndMessageAcked() {
        publisher.publishOrderCreated("1001", 250_000);

        await().atMost(Duration.ofSeconds(10)).untilAsserted(() -> {
            assertThat(shippingService.shippedOrders()).contains("1001");

            PendingMessagesSummary summary = redisTemplate.opsForStream()
                    .pending("orders", "shipping-group");
            assertThat(summary.getTotalPendingMessages()).isZero();
        });
    }
}

@ServiceConnection(name = "redis") 让 Spring Boot 自动从容器中获取主机和端口，所以你不需要 @DynamicPropertySource。await() 是必要的，因为 listener 运行在单独的线程上。断言 getTotalPendingMessages() 为 0 证明消息已被 ACK，而不仅仅是被读取。

9. 常见错误和何时不应使用

处理逻辑不是幂等的。 at-least-once 意味着一条消息可能被传递两次。例如：消费者处理完毕但在 ACK 前崩溃了。因此业务逻辑必须能够处理重复。常见的做法是检查 orderId 是否已处理过，或在数据库中使用唯一约束。

流无限增长。 ACK 不会从流中删除条目。你需要在写入时限制大小：

redisTemplate.opsForStream().add(record, XAddOptions.maxlen(100_000).approximateTrimming(true));

这个选项等同于 XADD orders MAXLEN ~ 100000 * ...。波浪号 ~ 允许 Redis 按内部节点修剪，比精确修剪快得多。

忘记 cancelOnError。 第一个异常会悄悄地停止消费者。症状是消息在流中堆积，但日志中没有明确的错误。

“幽灵”消费者。 每次用新的消费者名称部署时（例如 pod 名称变化），旧名称仍然留在消费者组中。你应该定期用 XGROUP DELCONSUMER 清理，但只在已认领了该消费者所有待处理消息之后。

何时不应使用 Redis Streams：

• 你需要存储很多天的事件并重放大量数据。Redis 将所有数据保存在 RAM 中，成本非常高。
• 你需要绝对的数据不丢失保证。使用 AOF everysec 或非同步副本故障转移，仍然可能丢失刚写入的一些消息。
• 吞吐量非常大，需要按数百个消费者分区和扩展。Kafka 更适合这种情况。

10. 总结

Redis Streams 结合消费者组为你提供了一个轻量级且可靠的事件队列，可以直接在你已有的 Redis 上运行。你需要记住四个关键点：

• 处理完毕后手动 ACK。
• 设置 cancelOnError(t -> false) 防止消费者悄悄停止。
• 有一个认领待处理消息的任务，并为 poison message 配备死信流。
• 编写幂等的处理逻辑。

添加 MAXLEN 来控制内存，并用 Testcontainers 编写集成测试，你就能为大多数 Spring Boot 应用构建一个足够可靠的 at-least-once 事件处理流程。