Spring Cloud Stream: giao tiếp bằng event qua RabbitMQ / Kafka

Ở các bài trước, các service trong spring-cloud-demo nói chuyện với nhau bằng HTTP qua OpenFeign. Bài này chuyển sang Event-Driven Architecture: dùng Spring Cloud Stream để giao tiếp qua RabbitMQ hoặc Kafka.

1. Bài toán: Order Service phải gọi quá nhiều service

Người dùng đặt hàng. Order Service cần:

Order Service
    |
    |---> Inventory Service      Reserve stock
    |
    |---> Payment Service        Process payment
    |
    |---> Notification Service   Send email

Nếu dùng OpenFeign:

@Service
public class OrderService {
    private final InventoryClient inventoryClient;
    private final PaymentClient paymentClient;
    private final NotificationClient notificationClient;

    public void createOrder(OrderRequest request) {
        inventoryClient.reserveStock(request);
        paymentClient.processPayment(request);
        notificationClient.sendEmail(request);
    }
}

2. Vấn đề

• Order Service phụ thuộc trực tiếp vào mọi downstream service.
• Một service chậm (ví dụ gửi email mất 5 giây) làm cả flow chậm theo.
• Thêm service mới (ví dụ loyalty-service cộng điểm) là phải sửa Order Service.
• Notification Service chết thì đơn hàng cũng lỗi, dù gửi email đâu phải việc bắt buộc.

Giải pháp: Order Service chỉ cần publish một event ORDER_CREATED. Ai quan tâm thì tự nghe.

Order Service
     |
     | ORDER_CREATED
     v
RabbitMQ / Kafka
     |
     +----> Inventory Service
     +----> Payment Service
     +----> Notification Service

3. Giải pháp: Spring Cloud Stream

Spring Cloud Stream là framework xây dựng message-driven microservices. Điểm quan trọng: nó là một lớp abstraction, code nghiệp vụ không phụ thuộc broker cụ thể.

// Spring AMQP: biết rõ exchange, routing key của RabbitMQ
rabbitTemplate.convertAndSend("order.exchange", "order.created", event);

// Spring Cloud Stream: chỉ biết tên binding
streamBridge.send("orderCreated-out-0", event);

Các khái niệm cần nhớ:

Từ Spring Cloud Stream 3.x, cách viết chuẩn là functional style, dùng đúng 3 functional interface của Java 8:

Khi cần gửi theo sự kiện nghiệp vụ (có người gọi API tạo đơn) thì không dùng Supplier mà dùng StreamBridge.

4. Code từng bước

Bước 1: Dependency (order-service, inventory-service, notification-service):

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>

Version được quản lý bởi spring-cloud-dependencies BOM như các bài trước.

Bước 2: Event và publisher trong order-service

public record OrderCreatedEvent(
        String eventId,
        String orderId,
        String productId,
        int quantity,
        String customerEmail) {}

@Service
public class OrderEventPublisher {
    private final StreamBridge streamBridge;

    public OrderEventPublisher(StreamBridge streamBridge) {
        this.streamBridge = streamBridge;
    }

    public void publish(OrderCreatedEvent event) {
        boolean sent = streamBridge.send("orderCreated-out-0", event);
        if (!sent) {
            throw new IllegalStateException("Failed to send order event");
        }
    }
}

Lưu ý: send() trả true không có nghĩa consumer đã xử lý xong. Muốn tin cậy end-to-end phải thêm ack, retry, idempotency và Outbox (phần câu hỏi phụ bên dưới).

Bước 3: Config order-service (đặt trong config-server repo, file order-service.yml):

spring:
  cloud:
    stream:
      bindings:
        orderCreated-out-0:
          destination: order.events
  rabbitmq:
    host: ${RABBITMQ_HOST:localhost}
    port: 5672
    username: ${RABBITMQ_USERNAME}
    password: ${RABBITMQ_PASSWORD}

Bước 4: Consumer trong inventory-service và notification-service

@Configuration
public class InventoryConsumerConfig {

    @Bean
    public Consumer<OrderCreatedEvent> orderCreated(InventoryService inventoryService) {
        return event -> inventoryService.reserve(event.orderId(),
                                                 event.productId(),
                                                 event.quantity());
    }
}

# inventory-service.yml
spring:
  cloud:
    function:
      definition: orderCreated
    stream:
      bindings:
        orderCreated-in-0:
          destination: order.events
          group: inventory-service

Notification Service giống hệt, chỉ khác group: notification-service và trong lambda gọi emailSender.sendOrderConfirmation(event).

Ví dụ Function: inventory nhận ORDER_CREATED rồi phát tiếp INVENTORY_RESERVED:

@Bean
public Function<OrderCreatedEvent, InventoryReservedEvent> reserveStock(InventoryService s) {
    return event -> s.reserve(event);   // giá trị trả về tự gửi sang reserveStock-out-0
}

spring:
  cloud:
    function:
      definition: reserveStock
    stream:
      bindings:
        reserveStock-in-0:  { destination: order.events,     group: inventory-service }
        reserveStock-out-0: { destination: inventory.events }

Bước 5: Test bằng curl

docker run -d --name rabbit -p 5672:5672 -p 15672:15672 rabbitmq:3-management

curl -X POST http://localhost:8080/api/orders \
  -H "Content-Type: application/json" \
  -d '{"productId":"P01","quantity":2,"customerEmail":"a@example.com"}'

# Log inventory-service:    Reserved 2 x P01 for order 1001
# Log notification-service: Sent confirmation email for order 1001

Mở RabbitMQ UI (cổng 15672) sẽ thấy exchange order.events và hai queue order.events.inventory-service, order.events.notification-service do binder tự tạo.

5. Cách hoạt động

order-service
     | streamBridge.send("orderCreated-out-0")
     v
Exchange: order.events  (topic exchange, binder tự tạo)
     |
     +-- binding --> Queue: order.events.inventory-service
     |                    |
     |                    +--> inventory #1  ┐ competing consumers:
     |                    +--> inventory #2  ┘ mỗi message chỉ 1 instance nhận
     |
     +-- binding --> Queue: order.events.notification-service
                          |
                          +--> notification #1

Hai ý quan trọng:

• Khác group = mỗi group nhận một bản sao (publish-subscribe).
• Cùng group = các instance chia nhau (competing consumers). Message có thể bị giao lại nếu xử lý lỗi, nên consumer phải idempotent.

Quên đặt group thì binder tạo queue anonymous, auto-delete: scale 3 instance sẽ thành 3 queue, mỗi instance xử lý trùng một lần, và message gửi lúc service tắt sẽ mất.

Với Kafka binder, cùng config trên map thành:

Topic: order.events  (3 partitions)
   P0 ──> consumer group "inventory-service" instance #1
   P1 ──> consumer group "inventory-service" instance #2
   P2 ──> consumer group "inventory-service" instance #2
   (group "notification-service" đọc lại toàn bộ topic với offset riêng)

6. So sánh

7. Khi nào nên dùng / không nên dùng

• Nên: một việc xảy ra mà nhiều service cần phản ứng (ORDER_CREATED); việc không cần kết quả ngay (gửi email, cập nhật thống kê); muốn tách rời service; triển khai Saga Choreography.
• Không nên: cần câu trả lời ngay để trả về user (kiểm tra giá, kiểm tra tồn kho trước khi hiển thị) thì dùng HTTP/Feign; hệ thống nhỏ, 1-2 service, thêm broker chỉ tăng vận hành; cần tính năng rất đặc thù của Kafka/Rabbit mà abstraction che mất.

Câu hỏi đúng chỗ. Mặc định Spring Cloud Stream retry trong bộ nhớ 3 lần (maxAttempts: 3) với backoff. Hết 3 lần mà vẫn lỗi thì với Rabbit binder, message bị reject và mất nếu chưa có DLQ. Vì vậy production luôn bật DLQ (Dead Letter Queue):

spring:
  cloud:
    stream:
      bindings:
        orderCreated-in-0:
          destination: order.events
          group: inventory-service
          consumer:
            max-attempts: 4
            back-off-initial-interval: 1000
            back-off-multiplier: 2.0
            back-off-max-interval: 10000
      rabbit:
        bindings:
          orderCreated-in-0:
            consumer:
              auto-bind-dlq: true
              republish-to-dlq: true   # kèm stack trace vào header

order.events.inventory-service
        |
        | xử lý lỗi: thử 1s, 2s, 4s ... hết max-attempts
        v
order.events.inventory-service.dlq   ← message nằm chờ, không mất
        |
        +--> alert, dev xem lỗi, sửa xong thì đẩy lại (shovel / script)

Với Kafka binder dùng enable-dlq: true và dlq-name: order.events.inventory.dlq trong spring.cloud.stream.kafka.bindings.<binding>.consumer.

Hai lưu ý:

• Chỉ retry lỗi tạm thời (timeout, DB sập). Lỗi dữ liệu (JSON sai, productId không tồn tại) retry cũng vô ích, nên khai báo vào retryable-exceptions với giá trị false để đi thẳng DLQ.
• Retry nghĩa là một message có thể được xử lý nhiều lần. Inventory nên lưu eventId đã xử lý (bảng processed_events) và bỏ qua nếu gặp lại.

Chưa an toàn, đây là bẫy hay gặp. @Transactional chỉ bao DB, không bao broker. Có thể xảy ra:

Order DB:  COMMITTED
RabbitMQ:  MESSAGE NOT PUBLISHED   (broker down, app crash giữa chừng)
=> Đơn có trong DB, inventory và notification không biết gì. Lệch trạng thái.

Đây là Dual Write Problem. Cách giải phổ biến: Transactional Outbox.

order-service
   | 1 local DB transaction
   +--> INSERT orders
   +--> INSERT outbox_events (payload ORDER_CREATED)
            | commit cùng lúc
            v
        Database
            |
            v
   Outbox Publisher (@Scheduled poll hoặc Debezium CDC)
            | streamBridge.send(...) rồi đánh dấu SENT
            v
        RabbitMQ / Kafka

@Transactional
public Order createOrder(OrderRequest req) {
    Order order = orderRepository.save(new Order(req));
    outboxRepository.save(OutboxEvent.of("ORDER_CREATED", toJson(order)));
    return order;
}

@Scheduled(fixedDelay = 1000)
public void relay() {
    for (OutboxEvent e : outboxRepository.findTop100ByStatusOrderByIdAsc("NEW")) {
        if (streamBridge.send("orderCreated-out-0", e.getPayload())) {
            e.markSent();
            outboxRepository.save(e);
        }
    }
}

Outbox vẫn có thể gửi trùng (gửi xong, crash trước khi đánh dấu SENT), nên lại quay về nguyên tắc: consumer phải idempotent.

Đúng rồi, đây chính là lợi ích của binder. Code Java (StreamBridge, Consumer<OrderCreatedEvent>) giữ nguyên. Chỉ đổi dependency và config:

<!-- thay spring-cloud-starter-stream-rabbit bằng -->
<artifactId>spring-cloud-starter-stream-kafka</artifactId>

spring:
  cloud:
    stream:
      kafka:
        binder:
          brokers: ${KAFKA_BROKERS:localhost:9092}
      bindings:
        orderCreated-out-0:
          destination: order.events
          producer:
            partition-key-expression: headers['orderId']   # cùng order vào cùng partition
            partition-count: 3

Nhưng chỉnh một điểm: "không sửa code" không có nghĩa "không cần suy nghĩ". Phần riêng của từng broker vẫn khác: DLQ config (rabbit.bindings... vs kafka.bindings...), thứ tự theo partition, khả năng replay của Kafka. Đổi binder là đổi hạ tầng, vẫn phải test lại.

Có thể dùng cả hai binder cùng lúc (multi-binder), ví dụ event nghiệp vụ qua Kafka, email qua RabbitMQ, bằng spring.cloud.stream.binders và bindings.xxx.binder: kafka.

Q1. Spring Cloud Stream là gì? Khác Spring Kafka / Spring AMQP thế nào?

🇻🇳 Tiếng Việt:

Spring Cloud Stream là framework để xây dựng microservice giao tiếp bằng message. Nó là lớp abstraction trên broker: em viết logic bằng Supplier, Function, Consumer hoặc gửi bằng StreamBridge, còn việc map sang exchange/queue của RabbitMQ hay topic của Kafka do binder lo qua config. Spring Kafka và Spring AMQP tích hợp trực tiếp với một broker, mạnh hơn khi cần tính năng đặc thù. Spring Cloud Stream gọn hơn và dễ đổi broker hơn.

🇬🇧 English:

Spring Cloud Stream is a framework for building message-driven microservices. It abstracts the broker: I write business logic as Supplier, Function or Consumer beans, or send with StreamBridge, and a binder maps bindings to RabbitMQ exchanges and queues or Kafka topics through configuration. Spring Kafka and Spring AMQP integrate directly with one broker and expose more broker-specific features, while Spring Cloud Stream is simpler and more portable.

Q2. Binder, binding, destination và consumer group là gì?

🇻🇳 Tiếng Việt:

Binder là cầu nối tới broker cụ thể. Binding là kênh vào/ra của app, đặt tên theo function như orderCreated-in-0. Destination là tên đích logic, ví dụ order.events, Rabbit map thành exchange, Kafka map thành topic. Consumer group: các instance cùng group chia nhau xử lý, mỗi message chỉ một instance nhận; các group khác nhau thì mỗi group nhận một bản sao. Trong project của em, inventory và notification là hai group nên cả hai đều nhận ORDER_CREATED.

🇬🇧 English:

A binder connects the app to a specific broker. A binding is an input or output channel named after the function, like orderCreated-in-0. A destination is the logical target, such as order.events, mapped to an exchange in RabbitMQ or a topic in Kafka. Instances in the same consumer group compete, so each message goes to one instance; different groups each get their own copy. In my project, inventory and notification are separate groups, so both receive ORDER_CREATED.

Q3. Consumer xử lý lỗi thì bạn làm gì?

🇻🇳 Tiếng Việt:

Em cấu hình retry với exponential backoff cho lỗi tạm thời, và bật DLQ để message lỗi hẳn được chuyển sang dead letter queue thay vì mất hoặc chặn queue. Lỗi dữ liệu thì không retry mà đưa thẳng vào DLQ. Em có alert khi DLQ có message để xử lý và replay. Vì retry có thể xử lý trùng nên consumer phải idempotent, ví dụ lưu eventId đã xử lý.

🇬🇧 English:

I configure retries with exponential backoff for transient errors and enable a DLQ so poison messages go to a dead letter queue instead of being lost or blocking the queue. Non-retryable errors, like invalid payloads, go straight to the DLQ. We alert on the DLQ and replay after fixing. Since retries can cause duplicates, consumers must be idempotent, for example by storing processed event IDs.

Q4. Làm sao đảm bảo event được gửi sau khi lưu DB?

🇻🇳 Tiếng Việt:

@Transactional không bao được cả DB và broker, nên có thể DB commit nhưng event không gửi được, gọi là dual write problem. Em dùng Transactional Outbox: lưu đơn hàng và bản ghi event vào bảng outbox trong cùng một transaction, rồi một publisher đọc outbox và gửi lên broker có retry. Outbox đảm bảo at-least-once, nên consumer vẫn phải idempotent.

🇬🇧 English:

@Transactional does not span the database and the broker, so the DB can commit while the event is never published. This is the dual write problem. I use the Transactional Outbox pattern: save the order and an outbox record in the same transaction, then a relay publishes outbox records to the broker with retries. That gives at-least-once delivery, so consumers still need to be idempotent.

Cách nhớ ngắn gọn

• Binder = cắm broker; binding = tên-in-0 / tên-out-0; destination = exchange/topic.
• Khác group = mỗi group một bản; cùng group = chia nhau.
• Quên group = queue anonymous, mất message, xử lý trùng.
• Lỗi: retry + backoff, rồi DLQ. Luôn idempotent.
• DB + event: Outbox, không tin @Transactional.