Pub/Sub 与 Redis Streams：区别与 Spring Boot 中的使用方法

1. 概述

Redis 提供两种不同的消息传递机制：Pub/Sub 和 Streams。许多刚使用 Redis 的开发者经常把这两者混淆，导致选择错误的工具，在系统重启时丢失数据。

本文解释了 Pub/Sub 和 Streams 之间的核心区别，为你提供可以在两者中运行的 Java 代码，并阐明何时应该使用哪一个。读完本文，你将能够自信地为你实际的消息传递问题选择正确的模式。

2. 什么是 Redis Pub/Sub

Pub/Sub（发布/订阅）按照“发送即忘”（fire-and-forget）的模式工作。发布者将消息发送到一个通道，Redis 立即将其转发给所有正在监听该通道的订阅者。如果没有订阅者在线，消息将永久消失。

2.1. Maven 依赖

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

2.2. 发布者

@Service
public class OrderEventPublisher {

    private final StringRedisTemplate redisTemplate;

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

    public void publish(String orderId, String status) {
        String message = "orderId=%s;status=%s".formatted(orderId, status);
        redisTemplate.convertAndSend("order-events", message);
    }
}

convertAndSend 将消息发送到 order-events 通道。如果还没有人订阅，这条命令仍然成功，但消息不会被保存。

2.3. 订阅者

@Component
public class OrderEventListener implements MessageListener {

    @Override
    public void onMessage(Message message, byte[] pattern) {
        String body = new String(message.getBody());
        System.out.println("接收事件: " + body);
    }
}

@Configuration
public class RedisPubSubConfig {

    @Bean
    public RedisMessageListenerContainer container(
            RedisConnectionFactory connectionFactory,
            OrderEventListener listener) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.addMessageListener(listener, new ChannelTopic("order-events"));
        return container;
    }
}

RedisMessageListenerContainer 管理专用连接（使用阻塞连接）来监听 order-events 通道。当发布者发送消息时，onMessage 会立即被调用。

发布事件时的输出：

接收事件: orderId=ORD-001;status=CONFIRMED

注意： 如果 OrderEventListener 还没有启动完成（container 还没有绑定），之前发布的消息将完全丢失，无法回放。

3. 什么是 Redis Streams

Stream 是一个有序的仅追加日志数据结构，每条消息都有唯一的 ID，格式为 <timestamp>-<sequence>。与 Pub/Sub 不同，Streams 在 Redis 中保存消息，直到你主动删除（XDEL）或裁剪（XTRIM）。消费者可以从头读取消息，按组读取（消费者组），具有类似 Kafka 的确认机制。

3.1. 将消息写入 Stream

@Service
public class OrderStreamPublisher {

    private final StringRedisTemplate redisTemplate;

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

    public RecordId publish(String orderId, String status) {
        Map<String, String> payload = Map.of(
                "orderId", orderId,
                "status", status
        );
        return redisTemplate.opsForStream()
                .add("order-stream", payload);
    }
}

opsForStream().add 将一条记录写入 order-stream 流中，并返回 RecordId —— 类似于记录的自生成主键。这条记录在 Redis 中存在，直到被裁剪或删除，即使还没有人读取。

3.2. 使用消费者组读取 Stream

@Component
public class OrderStreamConsumer {

    private final StringRedisTemplate redisTemplate;

    public OrderStreamConsumer(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
        createGroupIfAbsent();
    }

    private void createGroupIfAbsent() {
        try {
            redisTemplate.opsForStream()
                    .createGroup("order-stream", "order-processors");
        } catch (Exception e) {
            // Group 已存在，忽略
        }
    }

    public void pollOnce() {
        List<MapRecord<String, Object, Object>> records = redisTemplate.opsForStream()
                .read(Consumer.from("order-processors", "worker-1"),
                        StreamReadOptions.empty().count(10),
                        StreamOffset.create("order-stream", ReadOffset.lastConsumed()));

        for (MapRecord<String, Object, Object> record : records) {
            System.out.println("处理记录 " + record.getId() + ": " + record.getValue());
            redisTemplate.opsForStream().acknowledge("order-stream", "order-processors", record.getId());
        }
    }
}

createGroup 创建消费者组 order-processors，类似于 Kafka 中的消费者组。
Consumer.from("order-processors", "worker-1") 声明名为 worker-1 的消费者属于该组。同一组的多个工作者将分享记录，不会重复读取。
ReadOffset.lastConsumed() 表示读取从未分配给组内任何消费者的新记录。
acknowledge 确认已完成处理，Redis 从组的“待处理列表”中删除记录。

输出：

处理记录 1696500000000-0: {orderId=ORD-001, status=CONFIRMED}

注意： 如果你忘记调用 acknowledge，记录仍然会留在待处理列表中。其他消费者或同一消费者（重启后）可以使用 XCLAIM/readPending 重新获取处理 —— 这是 Pub/Sub 没有的“至少一次交付”机制。

4. Pub/Sub 和 Streams 的直接比较

标准	Pub/Sub	Streams
消息存储	否	是，直到裁剪/删除
重放旧消息	否	是（`XRANGE`、自定义偏移量）
消费者离线时丢失的消息	离线期间发送的所有消息都会丢失	不丢失，在线时重新读取
多消费者负载均衡	每个订阅者都收到副本（广播）	消费者组分担工作，每条记录由一个消费者处理
确认/重试	否	是（`XACK`、`XCLAIM`）
复杂程度	非常简单	更复杂，接近 Kafka
合适的用例	实时通知、无需保证	事件溯源、作业队列、需要保证处理

4.1. 何时使用 Pub/Sub

当你需要即时广播并且可以接受消息丢失，如果没有人在监听，就使用 Pub/Sub。典型的例子包括：更新聊天中的“正在输入…“状态、在应用的多个实例之间同步缓存（使缓存失效），或实时通知，即使错过几条也不重要。

@Service
public class CacheInvalidationPublisher {

    private final StringRedisTemplate redisTemplate;

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

    public void invalidate(String cacheKey) {
        redisTemplate.convertAndSend("cache-invalidation", cacheKey);
    }
}

每个应用实例订阅 cache-invalidation 通道，接收需要从本地缓存中删除的键。如果一个实例正在重启并错过事件，它只会在短时间内使用旧缓存 —— 这不是严重错误。

4.2. 何时使用 Streams

当消息是重要的业务数据，需要完全处理时，使用 Streams：订单、支付交易、审计日志。你需要确保消费者崩溃中途仍然可以在重启后重新处理该记录。

@Service
public class PaymentEventPublisher {

    private final StringRedisTemplate redisTemplate;

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

    public RecordId publish(String paymentId, BigDecimal amount) {
        Map<String, String> payload = Map.of(
                "paymentId", paymentId,
                "amount", amount.toString()
        );
        return redisTemplate.opsForStream().add("payment-stream", payload);
    }
}

每个支付事件都被记录到 payment-stream。即使 PaymentWorker 关闭以进行部署，该事件仍然在 Redis 中等待，工作者重启后将继续读取。

5. 为 Stream 消费者编写测试

因为 Streams 具有持久的状态，你应该仔细测试发布-消费-确认流程。使用 Testcontainers 运行真实的 Redis，避免模拟不正确的行为。

<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>junit-jupiter</artifactId>
    <scope>test</scope>
</dependency>

@Testcontainers
class OrderStreamConsumerTest {

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

    StringRedisTemplate redisTemplate;

    @BeforeEach
    void setUp() {
        RedisStandaloneConfiguration config = new RedisStandaloneConfiguration(
                redis.getHost(), redis.getFirstMappedPort());
        LettuceConnectionFactory factory = new LettuceConnectionFactory(config);
        factory.afterPropertiesSet();
        redisTemplate = new StringRedisTemplate(factory);
    }

    @Test
    void shouldConsumeAndAcknowledgeRecord() {
        OrderStreamPublisher publisher = new OrderStreamPublisher(redisTemplate);
        OrderStreamConsumer consumer = new OrderStreamConsumer(redisTemplate);

        publisher.publish("ORD-002", "SHIPPED");
        consumer.pollOnce();

        PendingMessages pending = redisTemplate.opsForStream()
                .pending("order-stream", Consumer.from("order-processors", "worker-1"),
                        Range.unbounded());

        assertThat(pending.getTotalPendingMessages()).isZero();
    }
}

此测试发布一条记录，调用 pollOnce() 来消费和确认，然后检查待处理列表是否为空 —— 这意味着记录已确认处理完成，不再挂起在组中。

6. 常见错误

为重要数据使用 Pub/Sub：许多初学者使用 Pub/Sub 来发送订单事件，然后在消费者重启时丢失事件感到惊讶。Pub/Sub 不是消息队列。
不裁剪 Stream：Stream 不会自动删除旧数据。如果你持续发布而不调用 XTRIM 或在 add 时设置 MAXLEN，Redis 内存会随时间增长。

redisTemplate.opsForStream().add(
        StreamRecords.newRecord()
                .in("order-stream")
                .ofMap(payload),
        RedisStreamCommands.XAddOptions.maxlen(10_000)
);

忘记在读取前创建消费者组：如果组不存在，XREADGROUP 命令会出错。始终使用 try-catch 调用 createGroup 来捕获 “BUSYGROUP”（组已存在）错误。
不为 Pub/Sub 设置连接超时：订阅者使用专用的阻塞连接；如果网络中断而没有重新连接机制，你将丢失从断开到重新连接之间的所有消息。

7. 结论

Pub/Sub 和 Streams 解决两个不同的问题，尽管都在 Redis 中。Pub/Sub 简单、快速，但不提供任何保证 —— 适合可以遗漏的实时通知。Streams 更复杂但更稳健，有消费者组和确认 —— 适合需要完全处理的业务数据，类似轻量级消息队列。

如果你的系统需要更高的可靠性（恰好一次、大分区、大规模并行多消费者组），考虑使用 Kafka 或 RabbitMQ，而不是强行让 Redis Streams 做太多事情。