多线程环境下的观察者：线程安全的事件分发、异步观察者模式

好，咱们今天聊点硬核的。观察者模式本身不难，但一旦放到多线程环境里，事情就变得有意思了。我在好几个高并发项目里都踩过这个坑——明明观察者模式写得挺漂亮，一上生产环境就出各种诡异问题。说白了，就是线程安全没处理好。

传统观察者模式的线程安全问题

先想想看，标准的观察者模式长什么样？一个Subject维护一个观察者列表，状态变化时挨个通知。这在单线程下没问题，但多线程并发时呢？

举个例子。你有一个事件源，两个线程同时往里面注册观察者。另一个线程正在遍历列表发通知。这时候会发生什么？

• ConcurrentModificationException：一个线程在遍历，另一个线程在修改列表
• 通知丢失：刚注册的观察者没收到本该收到的通知
• 死锁：观察者的回调方法里又去注册/注销其他观察者

嗯，这些都是我在项目中真实遇到过的。有一次线上告警，查了半天才发现是观察者列表的并发修改问题。

线程安全的事件分发策略

我个人习惯把线程安全的观察者模式分成三个层次来处理：

层次	策略	适用场景
基础	同步加锁	观察者少，通知快
进阶	读写分离	读多写少，通知频繁
高级	异步分发	观察者处理慢，或需要解耦

先看同步加锁的实现。最简单，但要注意锁的粒度：

public class ThreadSafeSubject {
    private final List<Observer> observers = new ArrayList<>();
    private final Lock lock = new ReentrantLock();

    public void register(Observer observer) {
        lock.lock();
        try {
            observers.add(observer);
        } finally {
            lock.unlock();
        }
    }

    public void notifyObservers(Event event) {
        List<Observer> snapshot;
        lock.lock();
        try {
            snapshot = new ArrayList<>(observers);
        } finally {
            lock.unlock();
        }
        // 在锁外遍历，避免死锁
        for (Observer observer : snapshot) {
            observer.onEvent(event);
        }
    }
}

这里有个关键点：我先把观察者列表快照出来，然后在锁外遍历。为什么？因为观察者的onEvent方法里可能又去调用register或unregister，如果还在锁内，就死锁了。我曾经在这个问题上栽过跟头，排查了好久。

异步观察者模式

说白了，同步通知有个硬伤：如果某个观察者处理得很慢，所有其他观察者都得等着。这在UI事件分发或者高并发场景下完全不能忍。

异步观察者模式的核心思路：事件分发线程和事件处理线程分离。我一般用事件队列+工作线程池来实现：

public class AsyncEventBus {
    private final ExecutorService executor;
    private final Map<Class<?>, List<EventHandler<?>>> handlers;
    private final ConcurrentLinkedQueue<Event> eventQueue;

    public AsyncEventBus(int threadCount) {
        this.executor = Executors.newFixedThreadPool(threadCount);
        this.handlers = new ConcurrentHashMap<>();
        this.eventQueue = new ConcurrentLinkedQueue<>();
        startDispatcher();
    }

    private void startDispatcher() {
        executor.submit(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                Event event = eventQueue.poll();
                if (event != null) {
                    dispatch(event);
                } else {
                    Thread.sleep(1); // 避免忙等
                }
            }
        });
    }

    @SuppressWarnings("unchecked")
    private void dispatch(Event event) {
        List<EventHandler<?>> matched = handlers.get(event.getClass());
        if (matched == null) return;

        for (EventHandler handler : matched) {
            executor.submit(() -> handler.handle(event));
        }
    }
}

核心知识体系

下面这张图是我自己总结的，把多线程观察者模式的核心脉络理清楚了：

避坑指南与最佳实践

我这些年积累了一些经验，分享给你：

• 不要阻塞事件分发线程：异步模式下，分发线程只负责投递事件，不负责处理。一旦分发线程被阻塞，整个系统就瘫痪了。
• 注意事件处理顺序：异步模式下，事件到达顺序和处理顺序不一定一致。如果业务上要求严格顺序，得用单线程池或者序列号机制。
• 处理回调中的异常：观察者的回调方法抛异常了怎么办？我的做法是捕获异常，记录日志，然后继续分发下一个。不能让一个观察者的异常影响整个系统。

实际项目中的取舍

我记得有个项目，需要实时推送行情数据到几百个客户端。一开始用的同步通知，结果一个客户端网络慢，所有客户端都跟着卡。后来改成异步，每个客户端独立线程处理，问题就解决了。

但异步也不是没有代价。你想想看，异步意味着事件处理延迟增加了，而且不好做事务回滚。所以我的建议是：

• 如果观察者处理很快（微秒级），同步加锁就够了
• 如果观察者处理慢（毫秒级以上），必须用异步
• 如果既要求顺序又要求异步，用单线程池+有界队列

嗯，说到底，架构设计就是取舍的艺术。没有完美的方案，只有最适合当前场景的方案。