Java中如何实现线程安全的事件发布订阅模式

P粉602998670 2025-10-28 00:00:00 次阅读

使用CopyOnWriteArrayList实现线程安全的事件发布订阅模式，1. 通过其读写分离特性保证并发读写时不抛出ConcurrentModificationException；2. 结合线程池异步通知提升性能并隔离异常；3. 监听器需自行同步内部状态；4. 可用弱引用避免内存泄漏。

在Java中实现线程安全的事件发布订阅模式，关键在于确保多个线程同时注册、注销或触发事件时不会出现并发问题。通常使用线程安全的数据结构来管理监听器，并保证事件通知过程的原子性和可见性。

使用CopyOnWriteArrayList管理监听器

最简单且高效的方式是使用CopyOnWriteArrayList存储订阅者列表。这个集合在修改时会复制底层数组，读操作无需加锁，适合读多写少的场景（如事件系统中频繁触发、较少增删监听器）。

示例代码：

import java.util.concurrent.CopyOnWriteArrayList;

public class EventBus {
    private final CopyOnWriteArrayList listeners = new CopyOnWriteArrayList<>();

    public void subscribe(EventListener listener) {
        listeners.add(listener);
    }

    public void unsubscribe(EventListener listener) {
        listeners.remove(listener);
    }

    public void publish(Event event) {
        for (EventListener listener : listeners) {
            listener.onEvent(event);
        }
    }
}

interface Event { }
interface EventListener {
    void onEvent(Event event);
}

优点：迭代过程中增删监听器不会抛出ConcurrentModificationException，天然支持并发读写。

避免长时间阻塞主线程

如果某些监听器处理耗时较长，同步通知可能影响性能。可通过异步方式提升响应性：

使用线程池执行回调，避免阻塞发布线程
注意异常隔离，个别监听器出错不应中断其他通知

改进publish方法：

private final ExecutorService executor = Executors.newCachedThreadPool();

public void publish(Event event) {
    for (EventListener listener : listeners) {
        executor.submit(() -> {
            try {
                listener.onEvent(event);
            } catch (Exception e) {
                // 记录日志，防止一个监听器崩溃影响整体
                System.err.println("Listener failed: " + e.getMessage());
            }
        });
    }
}