VI
FrançaisPortuguêsEspañolDeutschTiếng ViệtEnglish简体中文한국어日本語
Trang chủ / Kiến trúc & Thiết kế / Redis

Redis Streams với consumer group: xử lý sự kiện tin cậy

Cris The Dev · 12/10/2026 · 9 phút đọc · 109

1. Tổng quan

Pub/Sub của Redis có một nhược điểm lớn. Nếu consumer đang tắt lúc publish, message bị mất. Redis Streams giải quyết vấn đề này. Message được lưu bền trong stream, và mỗi consumer group ghi nhớ message nào đã được xử lý xong.

Bài này hướng dẫn bạn xây dựng luồng xử lý sự kiện đơn hàng với Spring Boot 3 và Spring Data Redis. Luồng gồm các bước: ghi sự kiện, đọc qua consumer group, ACK thủ công, nhận lại message bị treo và chuyển message lỗi sang dead-letter stream.

2. Redis Streams và consumer group hoạt động thế nào

Trước khi viết Java, bạn nên thử các lệnh gốc trong 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

Ý nghĩa từng lệnh:

  • XADD thêm một entry vào stream orders. Dấu * để Redis tự sinh ID dạng <millis>-<seq>.
  • XGROUP CREATE ... 0 tạo group shipping-group và bắt đầu đọc từ đầu stream. Nếu dùng $, group chỉ đọc message mới.
  • XREADGROUP ... > lấy message chưa giao cho ai trong group. Message được giao cho worker-1, đồng thời được ghi vào Pending Entries List (PEL) của group.
  • XPENDING cho biết có bao nhiêu message đã giao nhưng chưa ACK.
  • XACK xóa message khỏi PEL. Kể từ lúc này group coi message đã xử lý xong.

Output của XPENDING trước khi ACK trông như sau:

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

PEL là chìa khóa của độ tin cậy. Giả sử consumer crash sau khi nhận message nhưng trước khi ACK. Message vẫn nằm trong PEL, và một consumer khác có thể nhận lại nó (claim) để xử lý tiếp. Đây là ngữ nghĩa at-least-once.

3. Cài đặt

Thêm dependency Spring Data Redis. Starter này dùng Lettuce làm client mặc định:

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

Cấu hình kết nối và tên stream:

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

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

Tên consumer lấy từ HOSTNAME, nên mỗi instance (pod) có một tên riêng. Điều này quan trọng. Nếu hai instance dùng chung tên, Redis coi chúng là một consumer, và PEL của chúng bị lẫn vào nhau.

4. Ghi sự kiện vào stream

@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(...) tạo một MapRecord với các field dạng chuỗi. Lệnh opsForStream().add() tương đương XADD orders * ... và trả về RecordId do Redis sinh ra, ví dụ 1728640000000-0.

Dùng StringRedisTemplate giúp dữ liệu trong Redis đọc được bằng redis-cli. Bạn không phải đối mặt với chuỗi byte của JDK serialization.

5. Tạo consumer group an toàn

Group phải tồn tại trước khi đọc. Ứng dụng có thể khởi động khi stream chưa có message nào, nên bạn cần tham số MKSTREAM. Lệnh createGroup() của StreamOperations không bật tham số này. Vì vậy ta gọi xuống tầng connection:

@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;
            }
            // Group đã tồn tại: bình thường khi restart ứng dụng
        }
    }
}

Tham số cuối true bật MKSTREAM. Redis sẽ tạo stream rỗng nếu stream chưa có. Khi ứng dụng restart, Redis trả lỗi BUSYGROUP Consumer Group name already exists. Ta bỏ qua đúng lỗi này và ném lại mọi lỗi khác.

6. Đọc message với StreamMessageListenerContainer

6.1. Listener có ACK thủ công

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

Lệnh ACK chỉ chạy khi createShipment thành công. Nếu method ném exception, message vẫn nằm trong PEL để xử lý lại sau. Bạn đừng ACK trước khi xử lý. Làm vậy sẽ biến at-least-once thành at-most-once.

ShippingService trong bài là một service đơn giản. Nó ghi nhận các orderId đã tạo vận đơn:

@Service
public class ShippingService {

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

    public void createShipment(String orderId) {
        if (!shippedOrders.add(orderId)) {
            return; // đã xử lý rồi, bỏ qua (idempotent)
        }
        // gọi API đơn vị vận chuyển...
    }

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

6.2. Cấu hình container

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

Các điểm chính trong cấu hình:

  • ReadOffset.lastConsumed() tương ứng với > trong XREADGROUP, tức là chỉ lấy message mới của group.
  • pollTimeout là thời gian block của mỗi lần đọc. batchSize là số message tối đa trong một lần đọc.
  • autoAcknowledge(false) tắt ACK tự động, vì ta tự ACK trong listener.
  • cancelOnError(t -> false) là dòng quan trọng nhất. Mặc định, khi listener ném exception, container hủy subscription. Consumer sẽ âm thầm ngừng đọc mà không có cảnh báo rõ ràng.

Lưu ý: Bạn đừng dùng receiveAutoAck(...) cho nghiệp vụ quan trọng. Chế độ này ACK ngay khi đọc, tức là trước khi listener chạy xong. Nếu ứng dụng crash giữa chừng, message sẽ mất vĩnh viễn.

7. Xử lý message treo và dead-letter

Listener chỉ đọc message mới (>). Một message lỗi hoặc thuộc về consumer đã chết sẽ nằm mãi trong PEL nếu không ai xử lý nó. Ta cần một job định kỳ cho việc này:

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

Luồng xử lý của job:

  1. pending(...) tương đương XPENDING orders shipping-group - + 50. Mỗi PendingMessage cho biết consumer đang giữ message, thời gian idle và số lần đã giao.
  2. claim(...) tương đương XCLAIM. Lệnh này chuyển quyền sở hữu sang consumer hiện tại. Redis chỉ chuyển nếu message vẫn idle ít nhất MIN_IDLE, nhờ vậy hai instance chạy job cùng lúc không claim trùng nhau. Mỗi lần claim cũng tăng delivery count.
  3. Message giao quá MAX_DELIVERIES lần được chép sang stream orders:dlq rồi mới ACK. Message lỗi vĩnh viễn (poison message) vì thế không bị retry vô hạn.

Nếu createShipment ném exception, lệnh ACK không chạy. Message ở lại PEL và được thử lại ở chu kỳ sau. Nhớ bật @EnableScheduling trên một class @Configuration.

Lưu ý: Từ Redis 6.2 có lệnh XAUTOCLAIM. Lệnh này gộp bước XPENDING và XCLAIM làm một, và hiệu quả hơn khi PEL lớn. Bạn có thể gọi nó qua redisTemplate.execute(...) nếu phiên bản Spring Data Redis đang dùng chưa có API tiện lợi.

8. Viết test với Testcontainers

Thêm các dependency cho test:

<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>

Test dưới đây publish một sự kiện. Sau đó nó kiểm tra rằng vận đơn đã được tạo và PEL đã trống:

@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 tự lấy host và port từ container, nên bạn không cần @DynamicPropertySource. await() cần thiết vì listener chạy trên thread riêng. Assertion getTotalPendingMessages() bằng 0 chứng minh message đã được ACK chứ không chỉ được đọc.

9. Lỗi hay gặp và khi nào không nên dùng

Xử lý không idempotent. At-least-once nghĩa là một message có thể đến hai lần. Ví dụ: consumer xử lý xong nhưng crash trước khi ACK. Vì vậy nghiệp vụ phải chịu được việc lặp lại. Cách làm phổ biến là kiểm tra orderId đã xử lý chưa, hoặc dùng unique constraint trong database.

Stream phình to vô hạn. ACK không xóa entry khỏi stream. Bạn cần giới hạn kích thước khi ghi:

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

Tùy chọn này tương đương XADD orders MAXLEN ~ 100000 * .... Dấu ~ cho phép Redis cắt theo node nội bộ, nhanh hơn nhiều so với cắt chính xác.

Quên cancelOnError. Exception đầu tiên sẽ âm thầm dừng consumer. Triệu chứng là message dồn ứ trong stream mà log không có lỗi rõ ràng.

Consumer “ma”. Mỗi lần deploy với tên consumer mới (ví dụ tên pod), tên cũ vẫn còn trong group. Hãy định kỳ dọn bằng XGROUP DELCONSUMER, nhưng chỉ sau khi đã claim hết message pending của consumer đó.

Khi nào KHÔNG nên dùng Redis Streams:

  • Bạn cần lưu sự kiện nhiều ngày và replay khối lượng lớn. Redis giữ toàn bộ dữ liệu trong RAM, nên chi phí rất cao.
  • Bạn cần đảm bảo không mất dữ liệu tuyệt đối. Với AOF everysec hoặc failover replica bất đồng bộ, vẫn có thể mất một ít message vừa ghi.
  • Thông lượng rất lớn, cần partition và scale theo hàng trăm consumer. Kafka phù hợp hơn trong trường hợp này.

10. Kết luận

Redis Streams kết hợp với consumer group cho bạn một hàng đợi sự kiện nhẹ và tin cậy, chạy ngay trên Redis bạn đã có. Bạn cần nhớ bốn điểm chính:

  • ACK thủ công sau khi xử lý xong.
  • Đặt cancelOnError(t -> false) để consumer không âm thầm dừng.
  • Có job claim lại message treo, kèm dead-letter stream cho poison message.
  • Viết logic xử lý idempotent.

Thêm MAXLEN để kiểm soát bộ nhớ và test tích hợp với Testcontainers, bạn sẽ có một luồng xử lý sự kiện at-least-once đủ chắc chắn cho phần lớn ứng dụng Spring Boot.

redisredis-streamsspring-bootspring-data-redismessaging
Ứng dụng của mìnhAlterChụp ảnh thành tem, gom vào sổ temiOS · Android

Bài cùng chủ đề