Vì sao cần message queue: gửi email làm API chậm

Ok. Chương này dùng một project chạy xuyên suốt: shop-app. Trước khi học RabbitMQ có gì, cần hiểu vì sao phải đặt một message broker vào giữa các service.

1. Bài toán: đặt hàng xong phải gửi email, SMS, trừ kho

shop-app/
├── order-service          (nhận POST /orders)
├── notification-service   (gửi email / SMS)
└── inventory-service      (giữ hàng trong kho)

Phiên bản đầu tiên, order-service gọi thẳng các service khác bằng HTTP:

POST /orders
   |
order-service
   ├─ save order DB        100ms
   ├─ call inventory       200ms
   ├─ call notification    800ms  (gửi email qua SMTP provider)
   └─ call SMS             500ms
   |
response                ≈ 1600ms

2. Vấn đề

• Chậm: user chỉ cần biết "đặt hàng thành công", không cần chờ email gửi xong.
• Phụ thuộc chặt: email provider chết → notification-service lỗi → POST /orders lỗi theo, dù đơn hàng đã lưu.
• Khó mở rộng: thêm analytics-service là phải sửa code order-service để gọi thêm một API.
• Không chịu được đỉnh tải: flash sale 10.000 đơn/phút, email provider chỉ chịu 1.000 email/phút → timeout dây chuyền.

3. Giải pháp: message broker, xử lý bất đồng bộ

order-service chỉ lưu đơn rồi publish một message OrderPlaced vào broker. Các service khác tự lấy message ra xử lý khi rảnh.

POST /orders
   |
order-service
   ├─ save order DB             100ms
   └─ publish OrderPlaced         5ms
   |
response ≈ 105ms

           broker (RabbitMQ)
              |            |
   notification-service   inventory-service
   (gửi email sau)        (giữ hàng sau)

Lợi ích chính:

• Async: API trả về nhanh.
• Decoupling: producer không biết, không cần biết ai nghe.
• Buffer: consumer chậm thì message nằm chờ trong queue, không làm sập producer (load leveling).
• Chịu lỗi: notification-service chết 10 phút, message vẫn còn đó, khởi động lại thì xử lý tiếp.

4. Code từng bước

Bước 1: Dependency (cả producer và consumer)

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

Bước 2: Config. Chạy RabbitMQ local bằng Docker, credential lấy từ biến môi trường:

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 \
  -e RABBITMQ_DEFAULT_USER=$RABBIT_USER -e RABBITMQ_DEFAULT_PASS=$RABBIT_PASS \
  rabbitmq:4-management

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: ${RABBIT_USER}
    password: ${RABBIT_PASS}

Bước 3: Code Java. Event là một record, có eventId duy nhất (sẽ cần ở phần idempotency):

public record OrderPlaced(UUID eventId, Long orderId, Long userId,
                          BigDecimal total, Instant occurredAt) {}

@Service
@RequiredArgsConstructor
public class OrderService {

    private final OrderRepository orderRepository;
    private final RabbitTemplate rabbitTemplate;

    @Transactional
    public Order placeOrder(PlaceOrderRequest req) {
        Order order = orderRepository.save(Order.from(req));
        rabbitTemplate.convertAndSend("shop.events", "order.placed",
            new OrderPlaced(UUID.randomUUID(), order.getId(), order.getUserId(),
                            order.getTotal(), Instant.now()));
        return order;
    }
}

@Component
public class OrderPlacedListener {

    @RabbitListener(queues = "notification.order-placed")
    public void onOrderPlaced(OrderPlaced event) {
        emailSender.sendOrderConfirmation(event.userId(), event.orderId());
    }
}

Cần thêm Jackson2JsonMessageConverter làm bean để gửi JSON thay vì Java serialization. Exchange shop.events và queue sẽ khai báo ở bài 8.2.

Lưu ý: publish ngay trong @Transactional có lỗ hổng (DB rollback nhưng message đã đi, hoặc DB commit nhưng publish lỗi). Cách chuẩn là Transactional Outbox, học ở chương Distributed Systems.

Bước 4: Test

curl -X POST http://localhost:8081/orders \
  -H "Content-Type: application/json" \
  -d '{"userId":10,"items":[{"sku":"TSHIRT-01","qty":2}]}'
# trả về ngay ~100ms, email đến sau vài giây

# Tắt notification-service, đặt thêm đơn, mở Management UI :15672
# → queue notification.order-placed có Ready = 1. Bật lại → message được xử lý.

5. Cách hoạt động

Sync (HTTP)                         Async (message)
-----------                         ---------------
order ──call──> notification        order ──publish──> broker ──> queue
  |    chờ...                         | trả về ngay               |
  |<──200 / lỗi                       v                    consumer lấy ra
  v                                response                  xử lý, ACK
response (chậm, lỗi lan truyền)

Hai mô hình phân phối message:

Point-to-point (queue, competing consumers)
  queue ──> worker 1   msg1, msg3
        └─> worker 2   msg2, msg4      mỗi message chỉ 1 worker xử lý

Publish/Subscribe (pub/sub)
  OrderPlaced ──> notification  (bản sao 1)
              └─> inventory     (bản sao 2)   mỗi subscriber nhận một bản

6. So sánh sync và async

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

• Nên: việc phụ không cần kết quả ngay (email, SMS, thống kê), nhiều service cùng quan tâm một sự kiện, cần hấp thụ đỉnh tải, cần retry bền vững.
• Không nên: client cần kết quả ngay (kiểm tra tồn kho trước khi cho đặt, đăng nhập), hệ thống nhỏ một service, team chưa sẵn sàng xử lý eventual consistency và message trùng.

Được. Idempotency không chỉ dùng cho message queue, nó là concept nền của backend. Nhưng với queue thì gần như bắt buộc.

1. Idempotency là gì?

Một operation chạy nhiều lần nhưng kết quả cuối cùng giống như chạy một lần.

SET order.status = 'PAID'      chạy 1 lần hay 10 lần → PAID      idempotent
stock = stock - 2              chạy 2 lần → trừ 4                 KHÔNG idempotent
send email                     chạy 2 lần → 2 email               KHÔNG idempotent

2. Vì sao message bị giao trùng?

Broker chỉ xóa message khi nhận được ACK từ consumer:

broker ──OrderPlaced──> inventory-service
                          reserve stock: 10 → 8   ✅ commit DB
                          💥 crash trước khi ACK
broker: "chưa ACK" → giao lại cho instance khác
                          reserve stock: 8 → 6    ❌ trừ 2 lần

Đây là at-least-once delivery: broker đảm bảo không mất message, đổi lại có thể giao trùng. Ngoài crash, trùng còn đến từ producer retry khi không nhận được confirm, hoặc network chập chờn.

3. Giải pháp: idempotent consumer với bảng processed_messages

CREATE TABLE processed_messages (
    consumer     VARCHAR(100) NOT NULL,
    message_id   UUID         NOT NULL,
    processed_at TIMESTAMPTZ  NOT NULL DEFAULT now(),
    PRIMARY KEY (consumer, message_id)
);

Ý tưởng: đánh dấu "đã xử lý" và làm nghiệp vụ trong cùng một DB transaction. Unique key chặn lần thứ hai.

@Component
@RequiredArgsConstructor
public class InventoryListener {

    private final JdbcTemplate jdbc;
    private final InventoryService inventoryService;

    @RabbitListener(queues = "inventory.order-placed")
    @Transactional
    public void onOrderPlaced(OrderPlaced event) {
        int inserted = jdbc.update("""
            INSERT INTO processed_messages(consumer, message_id)
            VALUES ('inventory', ?) ON CONFLICT DO NOTHING
            """, event.eventId());
        if (inserted == 0) {
            return; // đã xử lý rồi → bỏ qua, vẫn ACK
        }
        inventoryService.reserve(event.orderId());
    }
}

Lần 1: INSERT ok (1 row) → reserve → COMMIT → ACK
Lần 2: INSERT conflict (0 row) → return → ACK, không trừ kho lần nữa
Crash giữa chừng lần 1: transaction rollback cả hai → lần 2 xử lý lại đúng

4. Các cách làm idempotent khác

Email là side effect bên ngoài DB, nên không thể bọc cùng transaction. notification-service lưu bảng notifications(id, status); gửi xong cập nhật SENT. Nếu provider gửi xong mà app chết trước khi cập nhật, vẫn có thể trùng. Đây là giới hạn thực tế: giảm trùng tối đa, không hứa exactly-once với hệ thống ngoài.

Queue là mỗi message chỉ một consumer nhận; pub/sub là mỗi bên quan tâm đều nhận một bản. Với RabbitMQ, pub/sub làm bằng cách cho exchange đẩy vào nhiều queue, "mỗi queue thuộc về một service", trong service đó có thể có nhiều instance chia nhau xử lý.

RabbitMQ                                   Kafka
                                           topic: order-events
exchange shop.events                         (log, giữ theo retention)
  ├─> queue notification.order-placed         ├─ group notification
  │     ├─ notification #1  ┐ chia nhau        │    ├─ instance #1 ┐ chia partition
  │     └─ notification #2  ┘                  │    └─ instance #2 ┘
  └─> queue inventory.order-placed            └─ group inventory
        └─ inventory #1                             └─ instance #1

Vậy đúng: khác biệt cốt lõi là RabbitMQ là queue (xử lý xong thì xóa), Kafka là log (giữ lại, ai đọc tới đâu tự nhớ offset). Chi tiết Kafka học ở bài 8.6.

Nếu muốn viết code không phụ thuộc broker nào, Spring có lớp trừu tượng Spring Cloud Stream. Chương này dùng Spring AMQP trực tiếp để thấy rõ cơ chế của RabbitMQ.

Q1. Vì sao dùng message queue?

🇻🇳 Tiếng Việt:

Message queue giúp xử lý bất đồng bộ và giảm coupling. Ví dụ đặt hàng: order-service lưu đơn rồi publish OrderPlaced, trả về ngay. Notification và inventory tự consume sau. Email provider chết thì message nằm chờ trong queue, không làm lỗi API đặt hàng. Queue còn đóng vai trò buffer khi có đỉnh tải. Đổi lại phải chấp nhận eventual consistency, thêm hạ tầng và xử lý message trùng.

🇬🇧 English:

A message queue enables asynchronous processing and decouples services. For example, order-service saves the order, publishes an OrderPlaced event and returns immediately; notification and inventory services consume it later. If the email provider is down, the message waits in the queue instead of failing the order API. The queue also buffers traffic spikes. The trade-offs are eventual consistency, extra infrastructure, and having to handle duplicate messages.

Q2. Idempotency là gì, sao consumer cần idempotent?

🇻🇳 Tiếng Việt:

Idempotency là chạy nhiều lần nhưng kết quả như chạy một lần. Broker thường giao at-least-once: consumer xử lý xong nhưng chết trước khi ACK thì message được giao lại. Vì vậy consumer phải idempotent. Em dùng message ID duy nhất, insert vào bảng processed_messages có unique key trong cùng transaction với nghiệp vụ; insert trùng thì bỏ qua.

🇬🇧 English:

Idempotency means running an operation many times has the same effect as running it once. Brokers usually deliver at-least-once: if a consumer finishes the work but crashes before acking, the message is redelivered. So consumers must be idempotent. I give each message a unique ID and insert it into a processed_messages table with a unique key, in the same transaction as the business change; a duplicate insert means the message is skipped.

Q3. Queue khác pub/sub thế nào?

🇻🇳 Tiếng Việt:

Queue là point-to-point: nhiều consumer cạnh tranh, mỗi message chỉ một consumer xử lý, dùng để chia tải. Pub/sub là mỗi subscriber nhận một bản sao. Trong RabbitMQ, pub/sub làm bằng exchange route vào nhiều queue, mỗi service một queue; trong Kafka là mỗi service một consumer group.

🇬🇧 English:

A queue is point-to-point: consumers compete and each message is processed by one of them, which spreads the load. Pub/sub gives every subscriber its own copy. In RabbitMQ we do pub/sub with an exchange routing to several queues, one per service; in Kafka each service uses its own consumer group.

Cách nhớ ngắn gọn

• Việc phụ không cần kết quả ngay → publish event, trả về sớm.
• Broker = at-least-once → consumer phải idempotent.
• processed_messages + unique key + cùng transaction với nghiệp vụ.
• Mỗi service một queue (bản sao); nhiều instance một queue (chia việc).