Event-Driven Architecture: Message, Event & Giao Tiếp Bất Đồng Bộ

OrderPlaced        ← Đơn hàng đã được đặt
PaymentProcessed   ← Thanh toán đã được xử lý
UserRegistered     ← Người dùng đã đăng ký
InventoryDepleted  ← Tồn kho đã hết

PlaceOrder         ← "Ê Order Service, đặt đơn hàng này"
ProcessPayment     ← "Ê Payment Service, thu tiền thẻ này"
SendEmail          ← "Ê Email Service, gửi email này"
ReserveInventory   ← "Ê Inventory Service, giữ hàng này"

GetOrderDetails    ← "Chi tiết đơn hàng 123 là gì?"
ListUserOrders     ← "User này có những đơn hàng nào?"
CheckInventory     ← "Còn bao nhiêu widget trong kho?"

{
  "type": "OrderPlaced",
  "orderId": "order-789",
  "timestamp": "2026-03-02T10:30:00Z"
}

{
  "type": "OrderPlaced",
  "orderId": "order-789",
  "userId": "user-123",
  "items": [
    {"productId": "prod-1", "name": "Widget", "quantity": 2, "price": 29.99},
    {"productId": "prod-2", "name": "Gadget", "quantity": 1, "price": 49.99}
  ],
  "totalAmount": 109.97,
  "shippingAddress": {
    "street": "123 Main St",
    "city": "Springfield",
    "zipCode": "62704"
  },
  "timestamp": "2026-03-02T10:30:00Z"
}

// Kafka producer — Spring Boot
@Service
public class OrderEventPublisher {
 
    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;
 
    public void publishOrderPlaced(Order order) {
        OrderPlacedEvent event = new OrderPlacedEvent(
            UUID.randomUUID().toString(),  // eventId
            order.getId(),
            order.getUserId(),
            order.getItems(),
            order.getTotalAmount(),
            Instant.now()
        );
 
        // Key = orderId → tất cả event cho cùng order vào cùng partition
        kafkaTemplate.send("order-events", order.getId(), event);
    }
}
 
// Kafka consumer — Spring Boot
@Component
public class PaymentEventConsumer {
 
    @KafkaListener(topics = "order-events", groupId = "payment-service")
    public void handleOrderPlaced(OrderPlacedEvent event) {
        log.info("Xử lý thanh toán cho order: {}", event.getOrderId());
        paymentService.processPayment(
            event.getUserId(),
            event.getTotalAmount(),
            event.getOrderId()
        );
    }
}

# Producer — thêm event vào stream
XADD order-events * type OrderPlaced orderId order-789 amount 109.97
 
# Consumer — đọc từ stream (consumer group)
XREADGROUP GROUP payment-service consumer-1 COUNT 10 BLOCK 5000 STREAMS order-events >

// Choreography — mỗi service phản ứng độc lập
 
// Order Service publish event
@Service
public class OrderService {
    public Order createOrder(CreateOrderCommand cmd) {
        Order order = orderRepository.save(Order.create(cmd));
        eventPublisher.publish(new OrderPlacedEvent(order));
        return order;
    }
 
    @EventListener
    public void onPaymentProcessed(PaymentProcessedEvent event) {
        Order order = orderRepository.findById(event.getOrderId());
        order.confirm();
        orderRepository.save(order);
        eventPublisher.publish(new OrderConfirmedEvent(order));
    }
}
 
// Inventory Service phản ứng với OrderPlaced
@Component
public class InventoryEventHandler {
    @KafkaListener(topics = "order-events")
    public void onOrderPlaced(OrderPlacedEvent event) {
        inventoryService.reserve(event.getItems());
        eventPublisher.publish(new InventoryReservedEvent(event.getOrderId()));
    }
}
 
// Payment Service phản ứng với InventoryReserved
@Component
public class PaymentEventHandler {
    @KafkaListener(topics = "inventory-events")
    public void onInventoryReserved(InventoryReservedEvent event) {
        paymentService.charge(event.getOrderId());
        eventPublisher.publish(new PaymentProcessedEvent(event.getOrderId()));
    }
}

// Orchestration — saga định nghĩa toàn bộ flow
@Component
public class CreateOrderSaga {
 
    public void execute(CreateOrderCommand command) {
        try {
            // Bước 1: Tạo order
            Order order = orderService.createOrder(command);
 
            // Bước 2: Giữ chỗ tồn kho
            inventoryService.reserve(order.getItems());
 
            // Bước 3: Xử lý thanh toán
            Payment payment = paymentService.charge(
                order.getUserId(), order.getTotalAmount()
            );
 
            // Bước 4: Xác nhận order
            orderService.confirm(order.getId(), payment.getId());
 
            // Bước 5: Gửi thông báo
            notificationService.sendOrderConfirmation(order);
 
        } catch (InventoryException e) {
            // Bù đắp: hủy order
            orderService.cancel(command.getOrderId());
            throw e;
 
        } catch (PaymentException e) {
            // Bù đắp: giải phóng inventory, hủy order
            inventoryService.release(command.getItems());
            orderService.cancel(command.getOrderId());
            throw e;
        }
    }
}

BEGIN TRANSACTION;
  UPDATE accounts SET balance = balance - 100 WHERE id = 1;
  UPDATE accounts SET balance = balance + 100 WHERE id = 2;
COMMIT;
-- Cả hai update thành công hoặc cả hai fail. Luôn nhất quán.

// Ví dụ compensation
public class CompensatingActions {
 
    // Action gốc                        // Compensation
    // createOrder()                  →   cancelOrder()
    // reserveInventory()             →   releaseInventory()
    // chargePayment()                →   refundPayment()
    // sendConfirmationEmail()        →   sendCancellationEmail()
    // awardLoyaltyPoints()           →   revokeLoyaltyPoints()
}

@KafkaListener(topics = "order-events")
public void handleOrderPlaced(OrderPlacedEvent event) {
    // Kiểm tra đã xử lý chưa
    if (processedEventRepository.existsById(event.getEventId())) {
        log.info("Event {} đã xử lý, bỏ qua", event.getEventId());
        return;
    }
 
    // Xử lý event
    paymentService.charge(event.getUserId(), event.getTotalAmount());
 
    // Đánh dấu đã xử lý (phải cùng transaction với side effect)
    processedEventRepository.save(new ProcessedEvent(event.getEventId()));
}

// ❌ KHÔNG idempotent — chạy hai lần cộng 100 hai lần
account.setBalance(account.getBalance() + 100);
 
// ✅ Idempotent — chạy hai lần cho cùng kết quả
account.setBalance(500); // Set giá trị tuyệt đối
 
// ✅ Idempotent — conditional update
UPDATE orders SET status = 'CONFIRMED'
WHERE id = ? AND status = 'PENDING';  -- Chỉ update nếu còn PENDING

-- Unique constraint ngăn xử lý trùng
CREATE TABLE processed_payments (
    order_id VARCHAR(255) PRIMARY KEY,  -- Không thể insert hai lần
    payment_id VARCHAR(255) NOT NULL,
    processed_at TIMESTAMP DEFAULT NOW()
);
 
-- Insert sẽ fail nếu trùng
INSERT INTO processed_payments (order_id, payment_id)
VALUES ('order-789', 'payment-456');

// Version 1: OrderPlaced event
{
  "orderId": "order-789",
  "amount": 109.97
}
 
// Version 2: Thêm field currency
{
  "orderId": "order-789",
  "amount": 109.97,
  "currency": "USD"          ← Field mới
}
 
// Version 3: Đổi tên amount → totalAmount
{
  "orderId": "order-789",
  "totalAmount": 109.97,     ← Đổi tên!
  "currency": "USD"
}

// Avro schema evolution — backward compatible
// Version 1
{
  "type": "record",
  "name": "OrderPlacedEvent",
  "fields": [
    {"name": "orderId", "type": "string"},
    {"name": "amount", "type": "double"}
  ]
}
 
// Version 2 — backward compatible (field mới có default)
{
  "type": "record",
  "name": "OrderPlacedEvent",
  "fields": [
    {"name": "orderId", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "currency", "type": "string", "default": "USD"}
  ]
}

// Xử lý nhiều version event
public void handleOrderPlaced(JsonNode event) {
    int version = event.has("schemaVersion")
        ? event.get("schemaVersion").asInt()
        : 1;  // Mặc định v1 cho event cũ
 
    switch (version) {
        case 1:
            processV1(event);  // Dùng "amount"
            break;
        case 2:
            processV2(event);  // Dùng "totalAmount" + "currency"
            break;
        default:
            log.warn("Schema version không biết: {}", version);
            processLatest(event);  // Cố gắng tốt nhất
    }
}

// Spring Kafka — Dead Letter Queue configuration
@Configuration
public class KafkaConfig {
 
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String>
            kafkaListenerContainerFactory() {
 
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
 
        // Retry 3 lần với backoff
        factory.setCommonErrorHandler(new DefaultErrorHandler(
            new DeadLetterPublishingRecoverer(kafkaTemplate),
            new FixedBackOff(1000L, 3)  // 1s delay, 3 retry
        ));
 
        return factory;
    }
}
 
// Consumer — failure tự động retry rồi gửi vào DLQ
@KafkaListener(topics = "order-events")
public void handleOrderEvent(OrderEvent event) {
    // Nếu throw 3 lần, message đi vào "order-events.DLT"
    processOrder(event);
}
 
// DLQ consumer — để điều tra và retry thủ công
@KafkaListener(topics = "order-events.DLT")
public void handleDeadLetter(ConsumerRecord<String, String> record) {
    log.error("Dead letter nhận được. Key: {}, Value: {}, Error: {}",
        record.key(), record.value(),
        new String(record.headers().lastHeader("kafka_dlt-exception-message").value())
    );
    alertService.notifyOncall("Dead letter trong order-events", record);
}

@KafkaListener(topics = "order-events")
public void handleOrderEvent(OrderEvent event) {
    try {
        processOrder(event);
    } catch (TransientException e) {
        // Throw lại — để retry mechanism xử lý
        throw e;
    } catch (ValidationException e) {
        // Poison message — log và skip (không retry)
        log.error("Event không hợp lệ, bỏ qua: {}", event, e);
        deadLetterService.send(event, e);
    } catch (InsufficientFundsException e) {
        // Business error — publish compensation event
        eventPublisher.publish(new PaymentFailedEvent(event.getOrderId(), e.getMessage()));
    }
}

// Base event class
public abstract class DomainEvent {
    private final String eventId;
    private final String eventType;
    private final Instant timestamp;
    private final int schemaVersion;
 
    protected DomainEvent(String eventType) {
        this.eventId = UUID.randomUUID().toString();
        this.eventType = eventType;
        this.timestamp = Instant.now();
        this.schemaVersion = 1;
    }
}
 
// Event cụ thể
public class OrderPlacedEvent extends DomainEvent {
    private final String orderId;
    private final String userId;
    private final List<OrderItem> items;
    private final BigDecimal totalAmount;
 
    public OrderPlacedEvent(String orderId, String userId,
                            List<OrderItem> items, BigDecimal totalAmount) {
        super("OrderPlaced");
        this.orderId = orderId;
        this.userId = userId;
        this.items = items;
        this.totalAmount = totalAmount;
    }
}
 
public class PaymentProcessedEvent extends DomainEvent {
    private final String orderId;
    private final String paymentId;
    private final BigDecimal amount;
    private final PaymentStatus status;  // SUCCESS, FAILED
 
    // constructor...
}
 
public class InventoryReservedEvent extends DomainEvent {
    private final String orderId;
    private final List<ReservedItem> reservedItems;
 
    // constructor...
}

@Service
public class OrderService {
 
    private final OrderRepository orderRepository;
    private final KafkaTemplate<String, DomainEvent> kafkaTemplate;
 
    public Order placeOrder(PlaceOrderRequest request) {
        // 1. Validate và tạo order locally
        Order order = Order.create(
            request.getUserId(),
            request.getItems()
        );
        order = orderRepository.save(order);
 
        // 2. Publish event — consumer phản ứng độc lập
        kafkaTemplate.send(
            "order-events",
            order.getId(),  // Partition key = orderId
            new OrderPlacedEvent(
                order.getId(),
                order.getUserId(),
                order.getItems(),
                order.getTotalAmount()
            )
        );
 
        log.info("Order đã đặt và event đã publish: {}", order.getId());
        return order;
    }
 
    @KafkaListener(topics = "payment-events", groupId = "order-service")
    public void onPaymentResult(PaymentProcessedEvent event) {
        Order order = orderRepository.findById(event.getOrderId())
            .orElseThrow();
 
        if (event.getStatus() == PaymentStatus.SUCCESS) {
            order.confirm(event.getPaymentId());
            log.info("Order đã xác nhận: {}", order.getId());
        } else {
            order.cancel("Thanh toán thất bại: " + event.getReason());
            log.warn("Order bị hủy do thanh toán thất bại: {}", order.getId());
        }
 
        orderRepository.save(order);
    }
}

@Service
public class PaymentEventHandler {
 
    private final PaymentService paymentService;
    private final ProcessedEventRepository processedEvents;
    private final KafkaTemplate<String, DomainEvent> kafkaTemplate;
 
    @KafkaListener(topics = "order-events", groupId = "payment-service")
    @Transactional
    public void onOrderPlaced(OrderPlacedEvent event) {
        // Kiểm tra idempotency
        if (processedEvents.existsById(event.getEventId())) {
            log.info("Event đã xử lý: {}", event.getEventId());
            return;
        }
 
        try {
            // Xử lý thanh toán
            Payment payment = paymentService.charge(
                event.getUserId(),
                event.getTotalAmount()
            );
 
            // Publish success event
            kafkaTemplate.send("payment-events", event.getOrderId(),
                new PaymentProcessedEvent(
                    event.getOrderId(),
                    payment.getId(),
                    event.getTotalAmount(),
                    PaymentStatus.SUCCESS
                )
            );
        } catch (PaymentException e) {
            // Publish failure event
            kafkaTemplate.send("payment-events", event.getOrderId(),
                new PaymentProcessedEvent(
                    event.getOrderId(),
                    null,
                    event.getTotalAmount(),
                    PaymentStatus.FAILED
                )
            );
        }
 
        // Đánh dấu event đã xử lý
        processedEvents.save(new ProcessedEvent(event.getEventId()));
    }
}

@Component
public class NotificationEventHandler {
 
    private final EmailService emailService;
    private final UserClient userClient;
 
    @KafkaListener(
        topics = {"order-events", "payment-events"},
        groupId = "notification-service"
    )
    public void handleEvent(DomainEvent event) {
        switch (event.getEventType()) {
            case "OrderPlaced" -> handleOrderPlaced((OrderPlacedEvent) event);
            case "PaymentProcessed" -> handlePayment((PaymentProcessedEvent) event);
        }
    }
 
    private void handleOrderPlaced(OrderPlacedEvent event) {
        User user = userClient.getUser(event.getUserId());
        emailService.send(
            user.getEmail(),
            "Đã Nhận Đơn Hàng",
            "Đơn hàng " + event.getOrderId() + " của bạn đã được nhận!"
        );
    }
 
    private void handlePayment(PaymentProcessedEvent event) {
        // Chỉ gửi email khi thành công — failure xử lý riêng
        if (event.getStatus() == PaymentStatus.SUCCESS) {
            // Lấy user info và gửi xác nhận...
        }
    }
}

# application.yml — Spring Boot Kafka config
spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all  # Chờ tất cả replica acknowledge
      retries: 3
      properties:
        enable.idempotence: true  # Kafka producer idempotency
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      auto-offset-reset: earliest
      enable-auto-commit: false  # Manual commit cho reliability
      properties:
        spring.json.trusted.packages: "com.example.events"

// ❌ Event soup — quá chi tiết
UserTableRowInserted
OrderValidationStarted
InventoryDatabaseQueryExecuted
CacheInvalidated
LogFileRotated

// ✅ Domain event có ý nghĩa
UserRegistered
OrderPlaced
PaymentProcessed
InventoryDepleted

// ❌ Over-engineering: async cho lookup đơn giản
User click "View Profile" → publish ViewProfileRequested event
→ Profile Service consume event → publish ProfileLoaded event
→ API Gateway consume event → trả về cho user
// Kết quả: 500ms latency thay vì 50ms, cộng thêm eventual consistency headache
 
// ✅ Thực dụng: sync cho read, async cho write
GET /users/123 → gọi sync trực tiếp → 50ms response
POST /orders → publish OrderPlaced event → "Đang xử lý..." response

// ❌ Chuỗi sync ngụy trang thành event
OrderService publish "PleaseValidateUser" event
→ chờ "UserValidated" event (blocking!)
→ rồi publish "PleaseCheckInventory" event
→ chờ "InventoryChecked" event (blocking!)

// Tuần 1: {orderId: "123", amount: 100}
// Tuần 2: {orderId: "123", totalAmount: 100}  ← Breaking change!
// Consumer cũ crash: field "amount" là null