Pub/Sub và Redis Streams: Khác biệt và Cách dùng với Spring Boot

1. Tổng quan

Redis cung cấp hai cơ chế truyền message khác nhau: Pub/Sub và Streams. Nhiều bạn mới dùng Redis thường nhầm hai cái này là một, dẫn đến chọn sai công cụ và mất dữ liệu khi hệ thống restart.

Bài này giải thích sự khác biệt cốt lõi giữa Pub/Sub và Streams, cho bạn code Java chạy được với cả hai, và nêu rõ khi nào nên dùng cái nào. Đọc xong, bạn sẽ tự tin chọn đúng mô hình cho bài toán nhắn tin thực tế của mình.

2. Redis Pub/Sub là gì

Pub/Sub (Publish/Subscribe) hoạt động theo mô hình “bắn và quên” (fire-and-forget). Publisher gửi message vào một channel, Redis chuyển tiếp ngay cho tất cả subscriber đang lắng nghe channel đó. Nếu không có subscriber nào online, message biến mất vĩnh viễn.

2.1. Dependency Maven

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

2.2. Publisher

@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 gửi message vào channel order-events. Nếu chưa có ai subscribe, lệnh này vẫn thành công nhưng message không được lưu lại đâu cả.

2.3. Subscriber

@Component
public class OrderEventListener implements MessageListener {

    @Override
    public void onMessage(Message message, byte[] pattern) {
        String body = new String(message.getBody());
        System.out.println("Nhận event: " + 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 quản lý kết nối riêng (dùng connection blocking) để nghe channel order-events. Khi publisher gửi message, onMessage được gọi ngay.

Đầu ra khi publish một event:

Nhận event: orderId=ORD-001;status=CONFIRMED

Lưu ý: Nếu OrderEventListener chưa khởi động xong (container chưa bind), message publish trước đó mất hoàn toàn, không có cách nào replay lại.

3. Redis Streams là gì

Stream là cấu trúc dữ liệu log có thứ tự, append-only, mỗi message có một ID duy nhất dạng <timestamp>-<sequence>. Khác với Pub/Sub, Streams lưu lại message trong Redis cho đến khi bạn chủ động xóa (XDEL) hoặc cắt bớt (XTRIM). Consumer có thể đọc lại từ đầu, đọc theo nhóm (consumer group) với cơ chế acknowledge giống Kafka.

3.1. Đưa message vào 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 ghi một record vào stream order-stream và trả về RecordId — giống primary key tự sinh của record đó. Record này tồn tại trong Redis cho tới khi bị trim hoặc xóa, dù chưa ai đọc.

3.2. Đọc Stream bằng Consumer Group

@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 đã tồn tại, bỏ qua
        }
    }

    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("Xử lý record " + record.getId() + ": " + record.getValue());
            redisTemplate.opsForStream().acknowledge("order-stream", "order-processors", record.getId());
        }
    }
}

• createGroup tạo consumer group order-processors, tương tự consumer group trong Kafka.
• Consumer.from("order-processors", "worker-1") khai báo consumer tên worker-1 thuộc group đó. Nhiều worker cùng group sẽ chia nhau record, không đọc trùng.
• ReadOffset.lastConsumed() nghĩa là đọc record mới chưa từng giao cho consumer nào trong group.
• acknowledge xác nhận đã xử lý xong, Redis xóa record khỏi “pending list” của group.

Đầu ra:

Xử lý record 1696500000000-0: {orderId=ORD-001, status=CONFIRMED}

Lưu ý: Nếu bạn quên gọi acknowledge, record vẫn nằm trong pending list. Consumer khác hoặc chính consumer này (sau khi restart) có thể dùng XCLAIM/readPending để lấy lại xử lý — đây là cơ chế “at-least-once delivery” mà Pub/Sub không có.

4. So sánh trực tiếp Pub/Sub và Streams

Tiêu chí	Pub/Sub	Streams
Lưu trữ message	Không	Có, đến khi trim/xóa
Replay lại message cũ	Không	Có (XRANGE, offset tùy chỉnh)
Consumer offline mất gì	Mất hết message gửi lúc offline	Không mất, đọc lại khi online
Load balancing nhiều consumer	Mỗi subscriber nhận bản riêng (broadcast)	Consumer group chia việc, mỗi record 1 consumer xử lý
Acknowledge / retry	Không có	Có (XACK, XCLAIM)
Độ phức tạp	Rất đơn giản	Phức tạp hơn, gần giống Kafka
Use case phù hợp	Thông báo real-time, không cần đảm bảo	Event sourcing, job queue, cần đảm bảo xử lý

4.1. Khi nào dùng Pub/Sub

Dùng Pub/Sub khi bạn cần broadcast tức thời và chấp nhận mất message nếu không ai đang nghe. Ví dụ điển hình: cập nhật trạng thái “đang gõ…” trong chat, đồng bộ cache giữa nhiều instance của ứng dụng (invalidate cache), hoặc thông báo real-time không quan trọng nếu bị bỏ lỡ một vài cái.

@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);
    }
}

Mỗi instance ứng dụng subscribe channel cache-invalidation, nhận key cần xóa khỏi local cache. Nếu một instance đang restart và bỏ lỡ event, nó chỉ dùng cache cũ trong một thời gian ngắn — không phải lỗi nghiêm trọng.

4.2. Khi nào dùng Streams

Dùng Streams khi message là dữ liệu nghiệp vụ quan trọng cần xử lý đầy đủ: đơn hàng, giao dịch thanh toán, log audit. Bạn cần đảm bảo consumer bị crash giữa chừng vẫn xử lý lại được record đó sau khi restart.

@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);
    }
}

Mỗi payment event được ghi vào payment-stream. Dù PaymentWorker đang tắt để deploy, event này vẫn chờ sẵn trong Redis, worker khởi động lại sẽ đọc tiếp.

5. Viết test cho Stream Consumer

Vì Streams có trạng thái bền (persistent), bạn nên test kỹ luồng publish – consume – acknowledge. Dùng Testcontainers để chạy Redis thật, tránh mock sai hành vi.

<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();
    }
}

Test này publish một record, gọi pollOnce() để consume và acknowledge, rồi kiểm tra pending list rỗng — nghĩa là record đã được xác nhận xử lý xong, không còn treo trong group.

6. Lỗi thường gặp

• Dùng Pub/Sub cho dữ liệu quan trọng: nhiều người mới dùng Pub/Sub để gửi event đơn hàng rồi ngạc nhiên khi consumer restart làm mất event. Pub/Sub không phải message queue.
• Không trim Stream: Stream không tự xóa dữ liệu cũ. Nếu bạn publish liên tục mà không gọi XTRIM hoặc đặt MAXLEN khi add, bộ nhớ Redis sẽ phình to theo thời gian.

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

• Quên tạo consumer group trước khi đọc: lệnh XREADGROUP sẽ lỗi nếu group chưa tồn tại. Luôn gọi createGroup với try-catch bắt lỗi “BUSYGROUP” (group đã có).
• Không set connection timeout cho Pub/Sub: subscriber dùng blocking connection riêng; nếu network đứt mà không có cơ chế reconnect, bạn mất hết message từ lúc đó đến khi phát hiện và reconnect.

7. Kết luận

Pub/Sub và Streams giải quyết hai bài toán khác nhau dù cùng nằm trong Redis. Pub/Sub đơn giản, nhanh, nhưng không đảm bảo gì — phù hợp cho thông báo tức thời có thể bỏ qua. Streams phức tạp hơn nhưng bền vững, có consumer group và acknowledge — phù hợp cho dữ liệu nghiệp vụ cần xử lý đầy đủ, giống một message queue nhẹ.

Nếu hệ thống của bạn cần độ tin cậy cao hơn nữa (exactly-once, phân vùng lớn, nhiều consumer group song song quy mô lớn), hãy cân nhắc Kafka hoặc RabbitMQ thay vì ép Redis Streams làm quá nhiều việc.