observer pattern

Giáo trình OOP hay minh họa Observer Pattern bằng trạm thời tiết (WeatherStation) cập nhật nhiệt độ lên màn hình. Nhưng đem Observer đồng bộ kiểu ngây thơ lên production của hệ thống tài chính hay thương mại điện tử thì rất dễ toang: một Observer gửi email bị timeout 30 giây sẽ giữ luôn thread của người dùng, hoặc một Observer ghi log bị lỗi sẽ rollback oan cả đơn hàng vừa thanh toán xong.

Observer Pattern ban đầu chỉ là chuyện gọi hàm trong bộ nhớ (in-process), về sau cùng ý tưởng đó lớn dần thành Event-Driven Architecture (EDA) và Reactive Streams.

Bài này nói về lý do JDK 9 deprecate java.util.Observable, cách gắn event với transaction bằng @TransactionalEventListener trong Spring Boot, và khi nào dùng event trong process là đủ, khi nào mới cần lên message broker (Kafka/RabbitMQ).

# các thế hệ tiến hóa của Observer pattern

# tại sao Java 9 lại deprecate Java.util.observable?

1. Khó mở rộng: Observable là một class chứ không phải interface. Java không cho đa kế thừa, nên class của bạn đã kế thừa class khác thì không thể làm Observable được nữa. Tệ hơn, setChanged() là protected, nên cũng không thể giữ một Observable bên trong rồi dùng (composition), bắt buộc phải kế thừa.
2. Mô hình sự kiện quá nghèo: chính Javadoc lúc deprecate ghi rằng thứ tự gửi thông báo không được đặc tả, và thay đổi trạng thái không tương ứng một-một với thông báo (gọi setChanged() mấy lần thì notifyObservers() cũng chỉ báo một lần). Các method của nó có synchronized, nhưng observer vẫn được gọi tuần tự ngay trên thread của người gọi notifyObservers(). Thêm nữa, danh sách observer giữ strong reference, quên huỷ đăng ký là rò rỉ bộ nhớ (chuyện này thì Observer tự viết tay cũng dính).
3. Không có kiểu: mọi thông báo chỉ truyền một tham số Object arg, observer phải tự ép kiểu, mất type-safety.

Javadoc gợi ý thay bằng java.beans (PropertyChangeListener) cho event model đầy đủ hơn, java.util.concurrent cho trao đổi giữa các thread, và Flow API cho kiểu reactive streams.

# cạm bẫy: ngoại lệ & quản lý giao dịch trong Spring

Khi một sự kiện nghiệp vụ xảy ra (ví dụ: OrderPlacedEvent), thường có nhiều observer cùng nghe:

• Observer 1: trừ tồn kho (Inventory, cần chung transaction với đơn hàng).
• Observer 2: gửi email xác nhận (Notification, gọi SMTP bên ngoài).
• Observer 3: tích điểm cho khách (Loyalty).

@EventListener mặc định chạy đồng bộ, trên chính thread và trong chính transaction của người gọi publishEvent(). Listener nào ném exception thì exception đó bay ngược về publishEvent() như một lời gọi hàm bình thường:

# giải pháp kiến trúc: @transactionaleventlistener kết hợp @async

Việc phụ (gửi mail, đẩy thông báo) không được phép ảnh hưởng tới transaction chính. Còn việc thật sự phải cùng sống cùng chết với đơn hàng như trừ tồn kho thì để @EventListener đồng bộ, cho nó rollback chung là đúng.

@Service
@RequiredArgsConstructor
public class OrderService {
 
    private final ApplicationEventPublisher eventPublisher;
    private final OrderRepository orderRepository;
 
    @Transactional
    public Order placeOrder(OrderRequest request) {
        Order order = orderRepository.save(new Order(request));
 
        // Listener AFTER_COMMIT chưa chạy ở đây, Spring giữ lại tới khi transaction commit
        eventPublisher.publishEvent(new OrderPlacedEvent(order.getId(), order.getTotalAmount()));
        return order;
    }
}

@Component
@RequiredArgsConstructor
public class OrderNotificationListeners {
 
    private static final Logger log = LoggerFactory.getLogger(OrderNotificationListeners.class);
    private final EmailClient emailClient;
 
    // KỸ THUẬT AN TOÀN: Chỉ chạy SAU KHI Transaction Database đã COMMIT thành công!
    @Async("notificationExecutor") // Chạy bất đồng bộ trên thread pool riêng, không chặn người dùng
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void onOrderPlaced(OrderPlacedEvent event) {
        log.info("Bắt đầu gửi email cho đơn hàng đã commit an toàn: {}", event.orderId());
        try {
            // Gọi SMTP Server bên ngoài
            emailClient.sendOrderConfirmation(event.orderId());
        } catch (Exception ex) {
            // Lỗi gửi mail chỉ ghi log, KHÔNG THỂ rollback được đơn hàng của khách!
            log.error("Không thể gửi email cho đơn hàng: {}", event.orderId(), ex);
        }
    }
}

@Async chỉ có tác dụng khi app có @EnableAsync. Ngoài ra còn vài chỗ hay vấp với @TransactionalEventListener:

• publishEvent() gọi ở chỗ không có transaction thì listener bị bỏ qua lặng lẽ, không lỗi gì cả. Muốn nó vẫn chạy thì đặt fallbackExecution = true.
• Listener AFTER_COMMIT chạy đồng bộ (không @Async) mà ghi DB thì transaction cũ đã commit xong, ghi tiếp sẽ không được commit nếu không mở transaction mới bằng @Transactional(propagation = Propagation.REQUIRES_NEW).
• Event chỉ nằm trong bộ nhớ. App chết ngay sau khi commit mà listener chưa chạy xong thì email đó mất luôn, không ai gửi lại. Việc nào bắt buộc phải xảy ra thì cần lưu event cùng transaction (Transactional Outbox, hoặc Event Publication Registry của Spring Modulith) rồi xử lý lại khi lỗi.

# tràn bộ đệm: backpressure & reactive streams

Chuyển observer sang bất đồng bộ thì lại sinh vấn đề mới: producer nhanh, consumer chậm.

Giả sử subject bắn ra 10.000 events/giây, trong khi observer chỉ xử lý được 500 events/giây:

• Hàng đợi bộ nhớ (In-Memory Queue) của Observer sẽ phình ra rất nhanh.
• Heap cạn và hệ thống sập với OutOfMemoryError.

# kiến trúc reactive streams (Java 9 flow API / project Reactor)

Observer truyền thống là push thuần: subject đẩy dữ liệu bất kể observer có chịu nổi hay không. Reactive Streams vẫn đẩy dữ liệu qua onNext(), nhưng chỉ đẩy trong phạm vi demand mà subscriber đã xin (người ta hay gọi là push-pull):

• Subscriber nhận một đối tượng Subscription.
• Subscriber gọi subscription.request(n), ý là "đưa tôi tối đa n phần tử nữa, xử lý xong tôi sẽ xin tiếp".
• Publisher không được gửi quá số đã xin, nên buffer phía consumer không phình vô hạn.

Có điều backpressure không làm dữ liệu thừa biến mất, nó chỉ đẩy câu hỏi "làm gì với phần thừa" ngược về phía nguồn. Nguồn kéo được (đọc DB, đọc file) thì chỉ cần đọc chậm lại. Nguồn không điều khiển được (click của người dùng, tick giá chứng khoán) thì vẫn phải chọn chiến lược: buffer có giới hạn, bỏ bớt, hoặc chỉ giữ giá trị mới nhất (onBackpressureBuffer, onBackpressureDrop, onBackpressureLatest trong Reactor).

# ma trận quyết định: in-process vs distributed message broker

Khi nào dùng Spring ApplicationEvents trong bộ nhớ, khi nào phải đưa lên Kafka / RabbitMQ?

# bảng so sánh chi tiết

# checklist trước khi vận hành

• 1. Bảo vệ transaction bằng AFTER_COMMIT: mọi việc phụ (gửi mail, gọi bên thứ ba, ghi log ra ngoài) nên dùng @TransactionalEventListener(phase = AFTER_COMMIT).
• 2. Thread pool riêng cho @Async: chỉ định rõ tên executor (ví dụ @Async("eventExecutor")), đặt queueCapacity có giới hạn và chọn rejection policy (vd CallerRunsPolicy). Executor mặc định của Spring Boot (applicationTaskExecutor) có queue gần như không giới hạn, Spring thuần không có Boot thì rơi về SimpleAsyncTaskExecutor tạo thread mới cho mỗi task; cả hai đều không chặn được tình trạng dồn việc.
• 3. Xử lý exception trong listener: exception từ method @Async kiểu void không về tới người gọi, mặc định chỉ được AsyncUncaughtExceptionHandler ghi log. Nên tự try-catch, ghi log đủ ngữ cảnh và quyết định retry hay bỏ qua.
• 4. Event phải immutable: dùng Java Record để listener này không sửa dữ liệu làm sai lệch listener khác. Record chỉ immutable ở bề mặt, field là List thì nên bọc bằng List.copyOf().
• 5. Theo dõi độ trễ xử lý: gắn metric đếm số event đang chờ trong queue của executor, để có cảnh báo khi hệ thống bắt đầu quá tải.

# lời kết

Observer Pattern không lỗi thời, nó chỉ đổi dạng. Cái ý tưởng giữ một danh sách observer rồi gọi lần lượt vẫn nằm bên dưới Spring Application Events, Reactive Streams và Event-Driven Architecture.

Khi dùng thì nắm cho rõ listener chạy đồng bộ hay bất đồng bộ, @TransactionalEventListener gắn với transaction thế nào, và event có được lưu lại hay không. Trong một service thì event in-process thường là đủ, chỉ khi cần đi qua nhiều service, cần lưu bền hay replay thì mới lên Kafka.

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 😎 👍🏻 🚀 🔥.