cqrs pattern

Bài này mình lấy ví dụ một hệ thống đặt vé xem phim (Movie Ticket Booking) có rất nhiều người vào cùng lúc, rồi xem CQRS (Command Query Responsibility Segregation), Saga Pattern và vài Design Patterns quen thuộc giải quyết mấy bài toán chính của nó ra sao.

# tính khả thi của cqrs

Hệ thống đặt vé xem phim lúc nào cũng có lượng đọc (Read) lớn hơn lượng ghi (Write) rất nhiều. Cả nghìn người có thể cùng lúc vào xem lịch chiếu và sơ đồ ghế, nhưng chỉ một phần nhỏ trong số đó thật sự giữ chỗ và thanh toán.

Chia theo từng Service thì mức độ hợp với CQRS như sau:

Kiểu Monolithic/CRUD truyền thống thì vướng gì?

Nếu dồn hết vào một TicketService ôm mọi thứ (God Service), bạn sẽ gặp:

• Database thành cổ chai: cùng một Transaction vừa phải lock row để giữ ghế (Pessimistic Locking), vừa phải join cả đống bảng để trả lịch chiếu.
• Nhìn code không ra nghiệp vụ: method kiểu updateTicketStatus(String id, int status) chẳng nói gì cả, status = 2 là đã thanh toán hay đã huỷ?
• Khó scale: không scale riêng được phần database cho đọc (hiện sơ đồ ghế) và phần cho ghi (thanh toán).

# cqrs thuần (lightweight)

Trước khi kéo cả một framework to đùng vào, cứ thử làm CQRS pattern thuần với Spring Boot và Java 21 đã. Đây là bước đệm rất ổn để dọn code cho gọn theo Clean Architecture.

# thiết lập command/query bus

Ý tưởng là tách riêng phần logic thay đổi trạng thái (Command) và phần logic truy vấn (Query).

# triển khai thực tế cho ticketing service

command model (đọc là hiểu nghiệp vụ): dùng record của Java 21 để khai báo các Command bất biến.

// Command: Rõ ràng, mang đậm ngôn ngữ nghiệp vụ (Ubiquitous Language)
public record ReserveSeatsCommand(
    UUID showtimeId,
    UUID userId,
    List<String> seatCodes
) {}
 
// Command Handler: Chỉ chứa logic xử lý thay đổi trạng thái
@Component
@RequiredArgsConstructor
public class ReserveSeatsCommandHandler {
    private final ShowtimeRepository writeRepository;
    private final ApplicationEventPublisher eventPublisher;
 
    @Transactional
    public ReservationResult handle(ReserveSeatsCommand cmd) {
        Showtime showtime = writeRepository.findByIdWithLock(cmd.showtimeId())
            .orElseThrow(() -> new ShowtimeNotFoundException(cmd.showtimeId()));
 
        // Domain logic: Giữ ghế và ném lỗi nếu ghế đã bị người khác đặt
        Reservation reservation = showtime.reserveSeats(cmd.userId(), cmd.seatCodes());
        writeRepository.save(showtime);
 
        // Publish Event để Read-side cập nhật lại sơ đồ ghế
        eventPublisher.publishEvent(new SeatsReservedEvent(
            reservation.getId(), cmd.showtimeId(), cmd.seatCodes(), reservation.getTotalAmount()
        ));
 
        return new ReservationResult(reservation.getId(), reservation.getExpiresAt());
    }
}

Event mang theo reservationId và totalAmount vì phần Saga bên dưới cần chúng để đi tiếp. Có một chỗ dễ vấp: ApplicationEventPublisher của Spring gọi listener đồng bộ, ngay trong transaction đang chạy. Nếu listener phía read cập nhật MongoDB bằng @EventListener thường mà sau đó transaction ghi bị rollback (vd lỗi lúc flush), thì sơ đồ ghế đã báo ghế bị giữ trong khi database ghi không có gì. Phía read nên nghe bằng @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT). Kể cả vậy, nếu ứng dụng chết ngay sau commit thì event vẫn mất, đó là lý do có Transactional Outbox ở phần sau.

query model (tối ưu cho đọc):

public record GetAvailableSeatsQuery(UUID showtimeId) {}
 
@Component
@RequiredArgsConstructor
public class GetAvailableSeatsQueryHandler {
    private final SeatMapProjectionRepository readRepository;
 
    @Cacheable(value = "seat-maps", key = "#query.showtimeId()")
    public SeatMapDTO handle(GetAvailableSeatsQuery query) {
        // Truy vấn thẳng vào MongoDB - Dữ liệu đã được denormalize, không cần JOIN
        return readRepository.findByShowtimeId(query.showtimeId())
            .orElseThrow(() -> new SeatMapNotFoundException(query.showtimeId()));
    }
}

@Cacheable ở đây cần đi kèm cách làm mới cache: sơ đồ ghế đổi liên tục, nên chỗ cập nhật projection (phần cuối bài) phải @CacheEvict key tương ứng, hoặc ít nhất đặt TTL thật ngắn. Nếu không, người dùng sẽ thấy ghế còn trống trong khi đã có người giữ. Đây là eventual consistency của CQRS: phía read luôn có thể trễ hơn phía write một chút, nên lúc giữ ghế thì phía write vẫn phải kiểm tra lại (như reserveSeats ở trên làm), không tin vào dữ liệu người dùng vừa nhìn thấy.

# kiến trúc axon framework & saga pattern

Khi một nghiệp vụ chạy qua nhiều microservices, quản lý transaction phân tán (Distributed Transaction) bắt đầu thành bài toán khó nhằn. Tới lúc này thì đưa Axon Framework vào để làm Event Sourcing và Saga Pattern là hợp lý.

# quản lý lifecycle đặt vé với saga

Quy trình đặt vé: Giữ chỗ (Ticketing) -> Thanh toán (Payment) -> Xuất vé (Ticketing) -> Gửi SMS (Notification).

Gọi API chéo nhau (Synchronous HTTP) thì một chỗ lỗi là kéo sập dây chuyền, nên ở đây mình dùng Orchestration-based Saga: một object Saga giữ trạng thái của cả quy trình, nghe event từng bước rồi gửi command cho bước kế tiếp hoặc command bù trừ khi có lỗi.

Code dưới viết theo API của Axon Framework 4.x. Lưu ý: event mà Saga nghe phải đi qua Event Bus của Axon (vd AggregateLifecycle.apply(...) trong aggregate), không phải ApplicationEventPublisher của Spring như handler ở phần trên.

@Saga
@Slf4j
public class TicketBookingSaga {
 
    @Autowired
    private transient CommandGateway commandGateway;
 
    // Bắt đầu quy trình khi ghế được giữ thành công
    @StartSaga
    @SagaEventHandler(associationProperty = "reservationId")
    public void on(SeatsReservedEvent event) {
        log.info("Ghế đã giữ, yêu cầu thanh toán cho Reservation: {}", event.reservationId());
        commandGateway.send(new CreatePaymentCommand(
            event.reservationId(), event.totalAmount()
        ));
    }
 
    // Xử lý khi thanh toán thành công
    @SagaEventHandler(associationProperty = "reservationId")
    public void on(PaymentCompletedEvent event) {
        log.info("Thanh toán thành công, tiến hành xuất vé");
        commandGateway.send(new IssueTicketCommand(event.reservationId()));
    }
 
    // Xử lý thất bại (Compensating Transaction)
    @SagaEventHandler(associationProperty = "reservationId")
    public void on(PaymentFailedEvent event) {
        log.warn("Thanh toán thất bại, nhả ghế đang giữ");
        commandGateway.send(new ReleaseSeatsCommand(event.reservationId()));
        SagaLifecycle.end(); // Quy trình đã kết thúc (thất bại), không để Saga sống mãi
    }
 
    // Kết thúc Saga khi vé đã xuất
    @EndSaga
    @SagaEventHandler(associationProperty = "reservationId")
    public void on(TicketIssuedEvent event) {
        log.info("Quy trình đặt vé hoàn tất: {}", event.ticketId());
    }
}

Vài điểm cần biết khi dùng Saga kiểu này:

• CommandGateway khai báo transient vì Axon serialize Saga xuống Saga Store sau mỗi event; resource được inject lại khi Saga được nạp lên.
• Saga phải kết thúc ở mọi nhánh. Nhánh thành công có @EndSaga, nhánh thất bại gọi SagaLifecycle.end(). Quên nhánh nào là Saga Store phình dần theo thời gian.
• Thiếu nhánh timeout: nếu bên Payment không bao giờ trả PaymentCompletedEvent hay PaymentFailedEvent (người dùng bỏ ngang trang thanh toán), ghế bị giữ mãi. Axon 4 xử lý bằng DeadlineManager: đặt deadline lúc bắt đầu Saga, hết hạn thì gửi ReleaseSeatsCommand.
• Command bù trừ như ReleaseSeatsCommand có thể bị gửi lại, nên phía nhận phải idempotent.
• Axon Framework 5 đã bỏ Saga (và DeadlineManager) khỏi core. @Saga, @SagaEventHandler... chỉ còn trong module legacy để hỗ trợ chuyển đổi; cách được khuyến nghị cho code mới là stateful event handler lưu trạng thái vào database. Xem Saga Migration.

# mấy design pattern hay dùng

# strategy pattern: tính giá vé linh hoạt

Giá vé đổi liên tục: vé cuối tuần, vé VIP, vé sinh viên, vé ngày lễ. Đừng viết một chuỗi if/else dài dằng dặc.

public interface PricingStrategy {
    boolean isApplicable(BookingContext context);
    BigDecimal calculate(BigDecimal basePrice, BookingContext context);
}
 
@Component
public class WeekendPricingStrategy implements PricingStrategy { /*...+20%...*/ }
 
@Component
public class StudentPricingStrategy implements PricingStrategy { /*...-15%...*/ }
 
// Trong BookingService
private final List<PricingStrategy> pricingStrategies; // Spring tự inject tất cả các bean implement interface này
 
public BigDecimal calculateFinalPrice(BigDecimal basePrice, BookingContext context) {
    return pricingStrategies.stream()
        .filter(strategy -> strategy.isApplicable(context))
        .findFirst()
        .map(strategy -> strategy.calculate(basePrice, context))
        .orElse(basePrice);
}

Hai chỗ cần để ý. Thứ tự bean trong List mà Spring inject không cố định nếu bạn không đánh @Order (hoặc implement Ordered), nên khi một vé khớp nhiều strategy (sinh viên đi cuối tuần) thì findFirst() có thể chọn strategy khác nhau giữa các lần deploy. Và findFirst() nghĩa là chỉ áp một quy tắc giá; nếu nghiệp vụ cho cộng dồn thì phải duyệt hết và áp lần lượt (reduce), đúng thứ tự mong muốn.

# transactional outbox pattern: giữ event nhất quán

Trường hợp ReserveSeatsCommand lưu database ngon lành nhưng gửi message qua Kafka lại lỗi (network timeout), thế là hai bên lệch dữ liệu.

Cách xử lý: lưu luôn event vào database RDBMS (PostgreSQL) trong cùng 1 Transaction. Sau đó một worker (Poller) đọc bảng outbox này và đẩy lên Kafka. Thay cho poller, có thể dùng CDC (vd Debezium đọc WAL của PostgreSQL) để đẩy bản ghi outbox lên Kafka.

@Transactional
public void reserveAndPublish(ReserveSeatsCommand cmd) {
    // 1. Cập nhật state vào nghiệp vụ
    showtimeRepository.save(showtime);
 
    // 2. Lưu event vào bảng Outbox cùng Transaction
    OutboxEvent event = OutboxEvent.builder()
        .aggregateId(showtime.getId())
        .eventType("SEAT_RESERVED")
        .payload(toJson(cmd))
        .build();
    outboxRepository.save(event);
}
 
private String toJson(Object value) {
    try {
        return objectMapper.writeValueAsString(value);
    } catch (JsonProcessingException e) {
        // Ném unchecked để @Transactional rollback cả state lẫn outbox
        throw new IllegalStateException("Không serialize được event", e);
    }
}
// Poller background sẽ đọc outboxRepository và push vào Kafka cluster, đảm bảo At-Least-Once Delivery.

Hàm toJson không chỉ để code biên dịch được (writeValueAsString của Jackson 2 ném checked JsonProcessingException). Mặc định @Transactional của Spring không rollback khi gặp checked exception. Nếu cứ khai báo throws JsonProcessingException cho qua, lúc serialize lỗi thì transaction vẫn commit phần state ghế mà không có bản ghi outbox, đúng cái lệch dữ liệu mà pattern này muốn tránh.

At-least-once nghĩa là event có thể tới hơn một lần (poller gửi xong nhưng chết trước khi đánh dấu đã gửi), nên consumer phải idempotent. Chi tiết xem bài transactional outbox và idempotent consumer.

# read model projection (materialized view)

Sơ đồ ghế trống là thứ bị query nhiều nhất hệ thống. Cứ join Room, Seat và Reservation liên tục trên phía write thì database ghi phải gánh thêm cả tải đọc, lúc cao điểm rất dễ quá tải.

Cách xử lý: lắng nghe các Domain Event (SeatsReservedEvent, SeatsReleasedEvent) rồi cập nhật thẳng vào một Document trên MongoDB.

@Document(collection = "showtime_seat_maps")
public class ShowtimeSeatMapProjection {
    @Id
    private UUID showtimeId;
    private String movieName;
    private List<SeatInfo> availableSeats;
    private List<SeatInfo> reservedSeats;
    // Dữ liệu đã được "nấu" sẵn, query API chỉ cần đọc một document theo _id rồi trả về JSON.
}

Phía cập nhật projection cũng có bẫy riêng: event có thể tới trùng hoặc sai thứ tự (SeatsReleasedEvent tới trước SeatsReservedEvent của cùng ghế). Cách thường dùng là lưu kèm version hoặc sequence của event cuối cùng đã áp dụng cho mỗi showtime, event nào cũ hơn thì bỏ qua. Và vì projection chỉ là dữ liệu dẫn xuất, nên có cách dựng lại nó từ đầu (replay event hoặc đọc lại từ database ghi) khi đổi cấu trúc hay khi nó bị lệch.

Chỉ là những ghi chép cá nhân với hy vọng mang lại chút giá trị. Nếu thấy hữu ích, đừng ngại chia sẻ cho bạn bè & đồng nghiệp nhé!

Happy coding 😎 👍🏻 🚀 🔥.