什么是观察者模式？应用场景有哪些？

结论先行：观察者（Observer）定义一对多依赖：主题（Subject）维护一份订阅者列表，状态变化时逐个回调，从而让"事件发生方"与"关心事件的多方"解耦。它的本质是回调注册 + 广播，而"发布/订阅"是在它之上再加一个中间人（Broker）的演进版。

1. 结构

                    ┌───────────────────────┐
                    │      Subject          │
                    │ observers: List<Obs>  │
                    ├───────────────────────┤
                    │ + attach(o)           │
                    │ + detach(o)           │
                    │ + notifyAll(msg) ─────┼──┐ 遍历回调
                    └───────────────────────┘  │
                                               │
                    ┌──────────────────────────┼──────────────┐
                    ▼                          ▼              ▼
            ┌───────────────┐         ┌───────────────┐ ┌───────────────┐
            │ SmsListener   │         │ PointListener │ │ MailListener  │
            │ onEvent(msg)  │         │ onEvent(msg)  │ │ onEvent(msg)  │
            └───────────────┘         └───────────────┘ └───────────────┘

关键约束：主题不认识具体监听器，只认接口；因此加一个"下单后发优惠券"的监听器，主题一行不改。

2. 观察者与发布/订阅的区别

一句话：观察者是"直接打电话"，发布订阅是"发到群里，谁在谁看"。

3. 代码：一个带异步与异常隔离的事件总线

import java.util.List;
import java.util.concurrent.*;

public class EventBus {
    public interface Listener {
        void onEvent(String topic, Object payload);
    }

    // 用 CopyOnWriteArrayList：通知时遍历，注册/注销时复制，避免 ConcurrentModificationException
    private final ConcurrentMap<String, List<Listener>> registry = new ConcurrentHashMap<>();
    private final ExecutorService pool = Executors.newFixedThreadPool(4);

    public void subscribe(String topic, Listener listener) {
        registry.computeIfAbsent(topic, k -> new CopyOnWriteArrayList<>()).add(listener);
    }

    public void unsubscribe(String topic, Listener listener) {
        List<Listener> list = registry.get(topic);
        if (list != null) {
            list.remove(listener);            // 忘记注销 = 内存泄漏
        }
    }

    public void publish(String topic, Object payload) {
        List<Listener> list = registry.getOrDefault(topic, List.of());
        for (Listener l : list) {
            try {
                pool.execute(() -> {
                    try {
                        l.onEvent(topic, payload);
                    } catch (Exception e) {
                        // 单个监听器异常不能影响其他监听器
                        System.err.println("listener failed: " + e.getMessage());
                    }
                });
            } catch (RejectedExecutionException e) {
                // 队列满时降级：记录并跳过，别阻塞发布方
                System.err.println("event dropped: " + topic);
            }
        }
    }
}

// 使用
EventBus bus = new EventBus();
bus.subscribe("order.paid", (t, p) -> System.out.println("发短信：" + p));
bus.subscribe("order.paid", (t, p) -> System.out.println("加积分：" + p));
bus.publish("order.paid", "ORDER-20260301-001");

三个工程要点：异步化（发布方不等）、异常隔离（每个监听器独立 try-catch）、注销（否则监听器被主题强引用，无法回收）。

4. 应用场景

5. 常见追问

Q：监听器里抛异常会怎样？ 同步实现下会中断后续通知，甚至把异常抛回发布方。所以要么每个回调 try-catch，要么统一交给异步线程池——上面代码两种都做了。

Q：为什么用 CopyOnWriteArrayList 而不是 ArrayList？ 通知过程中可能有监听器注销自己，普通 ArrayList 遍历会抛 ConcurrentModificationException。写少读多的场景正好适合 COW。

Q：观察者模式会内存泄漏吗？ 会。主题持有监听器的强引用，如果监听器是短生命周期对象（如某个请求的上下文），忘了 unsubscribe 就回收不掉。Spring 里对单例 Bean 监听没问题，但手动 addListener 的临时对象要显式移除。

Q：同步通知会不会拖慢主流程？ 会。下单主流程里同步发短信、发邮件，任何一环变慢都直接影响响应时间。做法：主流程只发事件，监听器进线程池或 MQ。

小结

• 观察者 = 注册回调 + 状态变更广播，解决"一对多通知"的解耦。
• 与发布订阅的分水岭是有没有中间人，以及是否跨进程、能否堆积。
• 工程上必须处理三件事：异步化、异常隔离、及时注销。
• 高频追问落在异常传播、并发修改与内存泄漏。