怎么实现异步观察者模式

异步观察者模式（Async Observer Pattern）的核心思想是：
当被观察者（Subject）状态变化时，通知观察者的过程是异步的，不阻塞主流程。

下面从 设计思路 → 常见实现方式 → 代码示例（Java / JS） → 注意事项 来系统说明。

一、核心设计思路

同步观察者模式的问题：

• notifyObservers() 会依次调用每个观察者
• 一旦某个观察者慢或阻塞，会影响整体性能

异步观察者模式的改进：

• 通知过程放入 异步线程 / 线程池 / 消息队列
• Subject 不等待 Observer 执行完成
• 观察者之间可以 并行或解耦执行

二、常见实现方式

✅ 方式 1：线程池 + 回调（最常用）

• Subject 拿到观察者列表
• 为每个观察者提交一个异步任务

✅ 方式 2：事件 / 消息队列

• Subject 发布事件
• Observer 订阅事件（如 MQ、EventBus）

✅ 方式 3：响应式编程（Reactive Streams）

• RxJava / Reactor / Observable

三、Java 示例（线程池实现）

1️⃣ 观察者接口

public interface Observer {
    void update(String event);
}

2️⃣ 被观察者（支持异步通知）

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

public class Subject {

    private final List<Observer> observers = new ArrayList<>();
    private final ExecutorService executor =
            Executors.newFixedThreadPool(4);

    public void addObserver(Observer observer) {
        observers.add(observer);
    }

    public void notifyObservers(String event) {
        for (Observer observer : observers) {
            executor.submit(() -> {
                try {
                    observer.update(event);
                } catch (Exception e) {
                    // 防止单个观察者异常影响其它任务
                    e.printStackTrace();
                }
            });
        }
    }

    public void shutdown() {
        executor.shutdown();
    }
}

3️⃣ 观察者实现

public class LogObserver implements Observer {
    @Override
    public void update(String event) {
        System.out.println("Log: " + event + " [" + Thread.currentThread().getName() + "]");
    }
}

public class EmailObserver implements Observer {
    @Override
    public void update(String event) {
        System.out.println("Send email: " + event);
    }
}

4️⃣ 使用

public class Main {
    public static void main(String[] args) {
        Subject subject = new Subject();

        subject.addObserver(new LogObserver());
        subject.addObserver(new EmailObserver());

        subject.notifyObservers("order_created");

        subject.shutdown();
    }
}

✅ 优点：

• 不阻塞主线程
• 观察者互不影响

四、JavaScript 示例（Promise / async）

1️⃣ 观察者

const observers = [];

function subscribe(observer) {
  observers.push(observer);
}

async function notifyObservers(data) {
  observers.forEach(async (observer) => {
    try {
      await observer(data);
    } catch (e) {
      console.error(e);
    }
  });
}

2️⃣ 使用

subscribe(async (msg) => {
  await new Promise(r => setTimeout(r, 1000));
  console.log("Log:", msg);
});

subscribe(async (msg) => {
  console.log("Email:", msg);
});

notifyObservers("order_created");

五、事件总线（更优雅）

Java（简化版 EventBus）

class EventBus {
    private final ExecutorService executor = Executors.newCachedThreadPool();
    private final Map<String, List<Runnable>> subscribers = new HashMap<>();

    public void subscribe(String event, Runnable task) {
        subscribers.computeIfAbsent(event, k -> new ArrayList<>()).add(task);
    }

    public void publish(String event) {
        subscribers.getOrDefault(event, List.of())
                   .forEach(executor::submit);
    }
}

六、注意事项（非常重要）

⚠️ 1. 异常必须捕获

否则异步线程可能直接终止

⚠️ 2. 顺序不保证

• 异步观察者 不保证执行顺序

⚠️ 3. 线程安全

• 观察者列表可能被并发修改

⚠️ 4. 资源释放

• 线程池要 shutdown()

七、什么时候用异步观察者？

✅ 适合：

• 日志、消息、通知
• 不需要返回结果
• 高并发、解耦系统

❌ 不适合：

• 强一致性
• 需要事务回滚
• 观察者之间有依赖顺序

八、总结一句话

异步观察者模式 = 观察者模式 + 异步执行（线程池 / 事件 / MQ）

如果你愿意，我可以：

• 给你一个 Spring Event 异步示例
• 或对比 观察者 vs 发布订阅
• 或用 Reactor / RxJava 实现响应式版本