Khi một đơn hàng chuyển sang trạng thái "đã thanh toán", cần trừ kho, gửi email hoá đơn, ghi log kiểm toán và báo cho hệ thống kế toán. Nếu OrderService.pay() gọi thẳng cả bốn việc, lớp này biết quá nhiều và phải sửa mỗi khi thêm một việc mới. Observer pattern giải quyết đúng chỗ đó: đối tượng phát sự kiện không cần biết ai đang lắng nghe, còn người lắng nghe tự đăng ký và tự rút lui. Bài này đi từ bản cài đặt thủ công, các lỗi hay gặp, tới những gì JDK và Spring cung cấp: PropertyChangeSupport, ApplicationEventPublisher và Flow API. Mọi đoạn code chỉ dùng cú pháp và API có sẵn trong Java 17 (đã kiểm tra bằng javac --release 17) và output được dán nguyên văn từ lần chạy thật.

Bài trước trong series, Strategy pattern với lambda và functional interface, đã cho thấy lambda giúp bỏ lớp con thế nào. Bài này dùng lại ý đó: hầu hết listener hiện đại chỉ là một lambda.

Observer pattern giải quyết vấn đề gì

Theo mô tả trên refactoring.guru, Observer pattern định nghĩa một cơ chế đăng ký để nhiều đối tượng được thông báo khi đối tượng mà chúng quan sát có sự kiện. Nó xử lý mâu thuẫn giữa hai phía: người nhận không nên liên tục hỏi "đã đổi chưa?", còn người phát không nên gửi tin cho những ai không quan tâm. Cuốn kinh điển Design Patterns của nhóm GoF gọi cặp vai trò này là Subject và Observer.

Vai tròNhiệm vụVí dụ trong bài
Subject (publisher)Giữ danh sách observer, cho đăng ký và huỷ, thông báo khi có sự kiệnOrderService, EventBus
Observer (subscriber)Khai báo một phương thức nhận thông báoConsumer<T>, PropertyChangeListener
EventDữ liệu đi kèm thông báorecord OrderPaid(String id)
ClientTạo subject và observer rồi nối chúng lạiHàm main hoặc cấu hình Spring

Lợi ích chính của Observer pattern là tuân thủ Open/Closed: thêm observer mới không sửa code của subject. Cái giá là thứ tự thông báo không được bảo đảm, và luồng điều khiển khó đọc hơn vì "ai sẽ chạy khi sự kiện này xảy ra" không còn nằm ở một chỗ.

Cài đặt Observer pattern bằng tay với lambda

Bản Observer pattern tối giản nhất chỉ cần một danh sách Consumer và hai thao tác. Phương thức subscribe trả về một Runnable để huỷ đăng ký, cách này đỡ phải giữ tham chiếu tới chính listener:

static class NaiveSubject<T> {
    private final List<Consumer<T>> listeners = new ArrayList<>();
    Runnable subscribe(Consumer<T> l) {
        listeners.add(l);
        return () -> listeners.remove(l);
    }
    void publish(T event) {
        for (Consumer<T> l : listeners) l.accept(event);
    }
}

Bản này chạy được trong demo một luồng, nhưng có ít nhất ba lỗi mà code thật sẽ gặp. Phần tiếp theo kiểm tra từng lỗi bằng chương trình Pitfalls.java.

Ba lỗi thường gặp khi tự cài Observer pattern

Lỗi 1 của Observer pattern: listener tự huỷ đăng ký trong lúc đang được gọi

Listener "một lần" (nhận sự kiện đầu tiên rồi rút) là nhu cầu rất phổ biến. Với ArrayList, người ta thường đoán sẽ gặp ConcurrentModificationException. Kết quả thực tế tinh vi hơn:

--- 1. Listener tự huỷ đăng ký khi đang được gọi (ArrayList) ---
  một-lần nhận A
--- 2. Cùng kịch bản với CopyOnWriteArrayList ---
  một-lần nhận A
  thường xuyên nhận A
  thường xuyên nhận B

Ở kịch bản 1 không có ngoại lệ nào, nhưng listener "thường xuyên" không nhận được sự kiện A. Lý do là sau khi phần tử đầu bị xoá, size() giảm từ 2 xuống 1 đúng bằng con trỏ của iterator, nên hasNext() trả false và vòng lặp kết thúc êm. Đây là loại lỗi nguy hiểm nhất vì không để lại dấu vết. Nếu danh sách có ba listener thì lần sau có thể ném ConcurrentModificationException. Cách sửa chuẩn là duyệt trên bản chụp (snapshot): CopyOnWriteArrayList cho iterator đọc từ mảng cũ nên thêm hay xoá giữa chừng không ảnh hưởng vòng lặp hiện tại, và cũng an toàn khi nhiều luồng đăng ký cùng lúc. Đổi lại, mỗi lần thêm hoặc xoá phải sao chép mảng, nên nó hợp với danh sách ít thay đổi và nhiều lần duyệt, đúng đặc điểm của danh sách listener.

Lỗi 2 của Observer pattern: một listener ném lỗi làm hỏng cả chuỗi

--- 3. Một listener ném lỗi ---
  publish ném hỏng; listener thứ hai không được gọi
  listener lỗi, bỏ qua: hỏng
  listener thứ hai nhận X

Nếu publish không bắt ngoại lệ, listener đầu hỏng sẽ cướp mất sự kiện của mọi listener sau nó, và lỗi còn lan ngược về nơi phát (ví dụ OrderService thấy ngoại lệ của module email). Bản SafeSubject bọc từng lần gọi trong try/catch rồi ghi log. Với Observer pattern, quyết định này là chính sách thiết kế, không phải chân lý: nếu listener là bước bắt buộc của giao dịch thì phải để lỗi nổi lên, còn nếu chỉ là thông báo phụ thì nên cô lập.

Lỗi 3 của Observer pattern: rò rỉ bộ nhớ vì quên huỷ đăng ký

Trong Observer pattern, subject giữ tham chiếu mạnh tới mọi observer, nên một observer ngắn hạn (ví dụ listener gắn với một màn hình hay một request) đăng ký vào subject sống lâu sẽ không bao giờ được giải phóng. Đây là "lapsed listener problem" kinh điển. Cách phòng: luôn trả về một đối tượng huỷ đăng ký như ở trên, dùng try-with-resources khi có thể, hoặc giữ observer bằng WeakReference nếu vòng đời không rõ ràng (đánh đổi là observer có thể biến mất khi chưa ai muốn).

Observer pattern trong JDK: từ Observable tới PropertyChangeSupport

JDK có cài sẵn Observer pattern từ bản 1.0 qua java.util.Observable và Observer, nhưng cả hai lớp này đã bị đánh dấu @Deprecated(since="9"). Tài liệu Javadoc của Observable nêu rõ lý do: mô hình sự kiện quá hạn chế, thứ tự 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. Javadoc đề xuất ba hướng thay thế: gói java.beans cho mô hình sự kiện phong phú hơn, các cấu trúc dữ liệu trong java.util.concurrent cho truyền thông điệp tin cậy giữa các luồng, và Flow API cho lập trình reactive streams. Vì Observable còn là một lớp (không phải interface), lớp nghiệp vụ của bạn bị buộc phải kế thừa nó, điều này cũng khó chịu khi lớp đã có cha.

Gói java.beans là lựa chọn nhẹ nhất khi cần theo dõi thuộc tính. PropertyChangeSupport lo việc quản lý danh sách listener, lọc theo tên thuộc tính và bỏ qua sự kiện khi giá trị cũ và mới bằng nhau:

static class Order {
    private final PropertyChangeSupport pcs = new PropertyChangeSupport(this);
    private String status = "NEW";

    void addListener(PropertyChangeListener l) { pcs.addPropertyChangeListener("status", l); }

    void setStatus(String s) {
        String old = this.status;
        this.status = s;
        pcs.firePropertyChange("status", old, s);   // không bắn nếu old.equals(new)
    }
}
status: NEW -> PAID
status: PAID -> SHIPPED

Lần gọi setStatus("PAID") thứ hai không sinh ra dòng nào, đúng như tài liệu mô tả. Swing, JavaBeans và nhiều thư viện giao diện dùng cơ chế này. Với code nghiệp vụ phía server, nó hơi thiên về "thuộc tính đổi" thay vì "sự kiện nghiệp vụ đã xảy ra", nên người ta thường chuyển sang event bus.

Event bus: biến thể của Observer pattern tách hẳn người phát

Ở dạng cơ bản, observer phải biết subject cụ thể để đăng ký. Event bus (một biến thể của Observer pattern, gần với Publish/Subscribe) thêm một trung gian: người phát chỉ đẩy sự kiện vào bus, người nghe đăng ký theo kiểu sự kiện. Hai bên không biết nhau. Dưới đây là một bus Observer pattern tối giản dùng sealed interface và record (đều có từ Java 17):

sealed interface OrderEvent permits OrderPlaced, OrderPaid {}
record OrderPlaced(String id, long amount) implements OrderEvent {}
record OrderPaid(String id) implements OrderEvent {}

static class EventBus {
    private final Map<Class<?>, List<Consumer<Object>>> handlers = new ConcurrentHashMap<>();

    <E> AutoCloseable on(Class<E> type, Consumer<? super E> handler) {
        Consumer<Object> wrapper = o -> handler.accept(type.cast(o));
        handlers.computeIfAbsent(type, k -> new CopyOnWriteArrayList<>()).add(wrapper);
        return () -> handlers.get(type).remove(wrapper);
    }

    void post(Object event) {
        for (var entry : handlers.entrySet()) {
            if (entry.getKey().isInstance(event)) {
                for (Consumer<Object> h : entry.getValue()) {
                    try { h.accept(event); }
                    catch (RuntimeException ex) { System.out.println("  handler lỗi: " + ex.getMessage()); }
                }
            }
        }
    }
}

Chạy với bốn handler (một nghe cả OrderEvent, hai nghe OrderPaid trong đó một cái cố tình hỏng), rồi huỷ handler audit:

post OrderPlaced:
  audit: OrderPlaced[id=DH-1, amount=500000]
  kho: giữ hàng cho DH-1
post OrderPaid:
  audit: OrderPaid[id=DH-1]
  mail: gửi hoá đơn DH-1
  handler lỗi: SMTP down
sau khi huỷ audit, post OrderPaid:
  mail: gửi hoá đơn DH-2
  handler lỗi: SMTP down

Handler audit đăng ký ở kiểu cha OrderEvent nên nhận cả hai loại sự kiện nhờ isInstance. Lỗi SMTP bị cô lập, handler mail trước đó vẫn chạy. Lưu ý: hai handler trong cùng một kiểu thì chạy theo thứ tự đăng ký, nhưng thứ tự giữa các kiểu khác nhau phụ thuộc vào ConcurrentHashMap, tức không nên dựa vào. Bus này chạy đồng bộ trên luồng của người phát. Các thư viện như Guava EventBus hay Spring cũng chọn mặc định đồng bộ vì dễ suy luận và giữ được ngữ cảnh giao dịch.

Observer pattern trong Spring: ApplicationEventPublisher và @EventListener

Spring đã có sẵn một event bus theo Observer pattern trong ApplicationContext. Theo tài liệu Spring Framework, mô hình này dựa trên ApplicationEvent và ApplicationListener, được gọi thẳng là "Observer design pattern chuẩn". Từ Spring 4.2 bạn có thể phát đi bất kỳ object nào (không bắt buộc kế thừa ApplicationEvent) và lắng nghe bằng annotation @EventListener. Ví dụ dùng Spring 6.0.11 (phát hành tháng 7/2023) với record làm sự kiện:

record OrderPaid(String orderId, long amount) {}

static class OrderService {
    private final ApplicationEventPublisher publisher;
    OrderService(ApplicationEventPublisher publisher) { this.publisher = publisher; }

    void pay(String orderId, long amount) {
        System.out.println("[" + Thread.currentThread().getName() + "] OrderService: ghi nhận thanh toán " + orderId);
        publisher.publishEvent(new OrderPaid(orderId, amount));
        System.out.println("[" + Thread.currentThread().getName() + "] OrderService: publishEvent đã trả về");
    }
}

static class MailListener {
    @EventListener @Order(2)
    void onPaid(OrderPaid e) { System.out.println("  MailListener nhận " + e); }
}

static class StockListener {
    @EventListener @Order(1)
    void onPaid(OrderPaid e) { System.out.println("  StockListener nhận " + e); }
}
[main] OrderService: ghi nhận thanh toán DH-1001
  StockListener nhận OrderPaid[orderId=DH-1001, amount=1250000]
  MailListener nhận OrderPaid[orderId=DH-1001, amount=1250000]
[main] OrderService: publishEvent đã trả về

Output xác nhận ba điều quan trọng. Thứ nhất, với Observer pattern kiểu Spring, OrderService không biết gì về hai listener. Thứ hai, listener chạy trên luồng main, cùng luồng với người phát, và publishEvent chỉ trả về sau khi cả hai chạy xong: đúng như tài liệu Spring ghi: "by default, event listeners receive events synchronously". Thứ ba, @Order quyết định thứ tự, nên StockListener (order 1) chạy trước MailListener dù được khai báo sau.

Vì đồng bộ, listener nằm trong cùng giao dịch của người phát. Điều này tiện (rollback kéo theo mọi việc) nhưng cũng nguy hiểm: gửi email trong listener đồng bộ rồi giao dịch rollback thì khách vẫn nhận hoá đơn của một đơn không tồn tại. Spring có @TransactionalEventListener (từ 4.2), mặc định chỉ chạy sau khi commit thành công (phase = AFTER_COMMIT), phù hợp cho các tác dụng phụ ra bên ngoài. Muốn chạy bất đồng bộ thì kết hợp @EventListener với @Async; tài liệu lưu ý ngoại lệ ở listener bất đồng bộ không truyền ngược về người phát, và listener loại này không thể trả về sự kiện kế tiếp.

Flow API: Observer pattern có back-pressure

Observer pattern truyền thống là "đẩy" (push): subject gửi bao nhiêu tuỳ ý, observer chậm thì dồn ứ hoặc làm nghẽn người phát. Java 9 thêm java.util.concurrent.Flow, bốn interface lồng nhau tương ứng đặc tả Reactive Streams. Theo Javadoc của Flow, Publisher sinh phần tử, Subscriber nhận qua onSubscribe, onNext, onError, onComplete, và Subscription nối hai bên với request(long n) để báo nhu cầu và cancel() để dừng. Điểm khác biệt cốt lõi: phần tử chỉ được giao khi subscriber đã yêu cầu. Đó là back-pressure.

Tiêu chíObserver pattern cổ điểnFlow API
Hướng điều khiểnSubject đẩy, observer không kiểm soát tốc độSubscriber chủ động request(n)
Kết thúc và lỗiKhông có tín hiệu chuẩnonComplete, onError
Thứ tựKhông đặc tảCác lời gọi tới một subscriber tuần tự
Chi phíRất thấpCao hơn: buffer, executor, vòng đời Subscription

JDK cung cấp sẵn SubmissionPublisher. Theo Javadoc, nó giao phần tử bất đồng bộ (mặc định qua ForkJoinPool.commonPool()), mỗi subscriber có buffer riêng với kích thước tối đa mặc định Flow.defaultBufferSize(), submit chặn người phát khi buffer đầy, còn offer thì bỏ phần tử. Subscriber dưới đây cố tình chậm (100 ms mỗi phần tử), chỉ xin một phần tử mỗi lần, và buffer chỉ bằng 4:

@Override public void onSubscribe(Flow.Subscription s) {
    this.sub = s;
    s.request(1);                       // chỉ xin 1 phần tử
}
@Override public void onNext(Integer item) {
    Thread.sleep(delayMs);              // xử lý chậm
    System.out.printf("  [%s] đã xử lý %d%n", ..., item);
    sub.request(1);                     // xử lý xong mới xin tiếp
}

Kết quả khi dùng submit (mười phần tử). Cột thời gian là thời điểm submit trả về, làm tròn tới 10 ms, đo trên máy 16 GB RAM (các mốc thời gian lệch nhau vài chục ms giữa các lần chạy):

Flow.defaultBufferSize() = 256
  producer submit(1) lúc +0 ms, lag ước lượng 1
  producer submit(2) lúc +0 ms, lag ước lượng 2
  producer submit(3) lúc +0 ms, lag ước lượng 3
  producer submit(4) lúc +0 ms, lag ước lượng 4
  producer submit(5) lúc +0 ms, lag ước lượng 5
  [fj-1] đã xử lý 1
  producer submit(6) lúc +100 ms, lag ước lượng 5
  [fj-1] đã xử lý 2
  producer submit(7) lúc +210 ms, lag ước lượng 5
  ...
  producer submit(10) lúc +520 ms, lag ước lượng 5
  ...
  hoàn tất

Năm phần tử đầu vào buffer ngay lập tức, rồi người phát bị chặn: mỗi lần submit tiếp theo chỉ trả về sau khi subscriber xử lý xong một phần tử (khoảng cách 100 ms khớp với thời gian xử lý). Đây chính là back-pressure: tốc độ của người phát bị kéo về tốc độ của người nhận. Con số 5 (không phải 4) là: 4 phần tử đang nằm trong buffer cộng 1 phần tử đang được subscriber xử lý trong onNext, vì request(1) chỉ được gọi sau khi xử lý xong nên phần tử đó chưa tính là đã tiêu thụ.

Khi chạy lại với dung lượng 3, 4 và 8, lag ước lượng lần lượt là 5, 5 và 9: dung lượng 3 được Javadoc cho phép làm tròn lên luỹ thừa của 2 (thành 4) nên cũng ra 5, còn 4 và 8 vốn đã là luỹ thừa của 2.

Lưu ý submit chỉ trả về một ước lượng độ trễ lớn nhất chứ không phải số đo chính xác, nên đừng dùng nó làm số đếm. Cùng kịch bản với offer thay submit:
  bỏ 5
  bỏ 7
  bỏ 8
  bỏ 9
  bỏ 10
  tổng số bị bỏ: 5
  [fj-1] đã xử lý 1
  [fj-1] đã xử lý 2
  [fj-1] đã xử lý 3
  [fj-1] đã xử lý 4
  [fj-1] đã xử lý 6
  hoàn tất

Người phát không chờ, năm phần tử bị loại và subscriber chỉ nhận phần còn lại (ở lần chạy này là 1, 2, 3, 4 và 6). Phần tử nào bị bỏ phụ thuộc vào thời điểm subscriber rút phần tử đầu khỏi buffer, nên không đoán trước được: lần này phần tử 6 lọt vào vì subscriber vừa lấy xong phần tử 1, còn phần tử 5 thì không. Chọn submit hay offer là chọn giữa "không mất dữ liệu" và "không bao giờ chặn": thanh toán thì chọn cái trước, còn số liệu hiển thị thời gian thực có thể chọn cái sau. Với pipeline phức tạp (map, filter, gộp luồng), các thư viện như Project Reactor hay RxJava cài đặt sẵn toán tử trên đặc tả này; bản thân JDK chỉ cung cấp bốn interface và SubmissionPublisher.

Khi nào không nên dùng Observer pattern

Observer pattern không phải lúc nào cũng đúng. Các trường hợp sau nên cân nhắc cách khác:

  • Luồng nghiệp vụ bắt buộc theo thứ tự. Nếu "trừ kho" phải xong trước "tạo vận đơn", hãy gọi tường minh. Observer pattern làm thứ tự thành ngầm định và khó kiểm thử.
  • Chỉ có một bên nghe, và mãi mãi chỉ một. Gọi trực tiếp rõ ràng hơn nhiều. Đừng dựng bus cho hai class.
  • Cần kết quả trả về. Observer pattern là một chiều; nếu người phát cần câu trả lời, hãy dùng interface thường hoặc Strategy.
  • Tác dụng phụ ra bên ngoài trong giao dịch. Gửi email, gọi API bên thứ ba trong listener đồng bộ có thể vượt ngoài phạm vi rollback. Dùng AFTER_COMMIT hoặc hàng đợi bền vững (outbox) thay vì tin vào sự kiện trong bộ nhớ.
  • Sự kiện phải sống sót qua restart. Event bus trong bộ nhớ mất hết khi tiến trình chết. Khi đó cần message broker, không phải Observer pattern.

Checklist khi cài đặt hoặc review Observer pattern

  • Với Observer pattern, duyệt danh sách listener trên bản chụp (CopyOnWriteArrayList hoặc sao chép trước khi lặp).
  • Quyết định rõ chính sách lỗi: cô lập từng listener hay để ngoại lệ nổi lên.
  • Luôn có cách huỷ đăng ký, và gọi nó khi observer hết vòng đời.
  • Ghi rõ sự kiện chạy đồng bộ hay bất đồng bộ, trên luồng nào, trong giao dịch nào.
  • Không dựa vào thứ tự listener trừ khi có cơ chế rõ ràng như @Order.
  • Trong Observer pattern, dùng event bất biến (record) để listener không sửa dữ liệu của nhau.
  • Với producer nhanh hơn consumer, chọn Flow và quyết định submit hay offer.
  • Không dùng java.util.Observable trong code mới.

Tóm tắt

Observer pattern tách người phát khỏi người nghe, và cùng một ý tưởng xuất hiện ở nhiều mức: danh sách Consumer tự viết, PropertyChangeSupport trong JDK, ApplicationEventPublisher trong Spring và Flow API có back-pressure. Phần khó không nằm ở cấu trúc mà ở những quyết định xung quanh: đồng bộ hay không, lỗi xử lý ra sao, ai huỷ đăng ký, giao dịch kết thúc lúc nào. Các số đo trong bài lấy từ lần chạy thật; mã nguồn nằm trong code/s6-13/.

Bài tiếp theo

Observer pattern cho phép gắn thêm hành vi từ bên ngoài. Bài kế tiếp xem hướng ngược lại, khi lớp cha giữ khung thuật toán và lớp con chỉ điền các bước: Template Method: khung thuật toán và điểm mở rộng. Nếu muốn xem lại cách tổ chức đối tượng dạng cây trước đó, đọc Composite pattern: xử lý cây đối tượng đồng nhất.