38. 设计模式面试题3：实现一个观察者模式，支持异步通知

观察者模式，说白了就是「发布-订阅」那一套。一个对象状态变了，通知一堆依赖它的对象。这个模式本身不难，但一旦加上「异步」两个字，坑就来了。

我记得有一次面试，候选人把观察者模式背得滚瓜烂熟，代码写得也漂亮。我一问「如果通知是异步的，你怎么设计？」他愣了一下，然后说「那就开线程呗」。嗯，开线程确实是个办法，但怎么开、开多少、线程安全怎么保证、回调里能不能加锁……这些才是面试官真正想听的。

今天我们就来聊聊，怎么实现一个支持异步通知的观察者模式。我会从基础讲起，然后一步步加上异步能力。

基础版：同步观察者

先看一个最朴素的实现。我习惯用模板加函数对象，这样灵活性高一些。

#include <functional>
#include <vector>
#include <algorithm>
#include <mutex>

template<typename... Args>
class Subject {
public:
    using Observer = std::function<void(Args...)>;

    void attach(Observer obs) {
        std::lock_guard<std::mutex> lock(m_mutex);
        m_observers.push_back(std::move(obs));
    }

    void detach(const Observer& obs) {
        std::lock_guard<std::mutex> lock(m_mutex);
        // 这里需要比较函数对象，实际项目中通常用句柄或ID
        // 此处省略具体实现
    }

    void notify(Args... args) {
        std::lock_guard<std::mutex> lock(m_mutex);
        for (auto& obs : m_observers) {
            obs(args...);
        }
    }

private:
    std::vector<Observer> m_observers;
    std::mutex m_mutex;
};

这个版本的问题很明显：notify 是同步的。如果某个观察者的回调很慢，整个通知链都会被阻塞。你想想看，一个 UI 事件触发了，结果因为某个日志观察者写文件太慢，界面卡住了——这谁受得了？

异步版：引入线程池

要支持异步通知，最简单的思路就是每个观察者开一个独立线程去执行。但这样做太粗暴了，线程创建销毁的开销很大，而且不好控制并发数量。

我个人更推荐用线程池。把通知任务丢进线程池，让线程池去调度。

#include <functional>
#include <vector>
#include <memory>
#include <future>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>

class ThreadPool {
public:
    explicit ThreadPool(size_t threads) : m_stop(false) {
        for (size_t i = 0; i < threads; ++i) {
            m_workers.emplace_back([this] {
                while (true) {
                    std::function<void()> task;
                    {
                        std::unique_lock<std::mutex> lock(m_mutex);
                        m_cv.wait(lock, [this] {
                            return m_stop || !m_tasks.empty();
                        });
                        if (m_stop && m_tasks.empty()) return;
                        task = std::move(m_tasks.front());
                        m_tasks.pop();
                    }
                    task();
                }
            });
        }
    }

    template<typename F>
    auto enqueue(F&& f) -> std::future<decltype(f())> {
        using return_type = decltype(f());
        auto task = std::make_shared<std::packaged_task<return_type()>>(
            std::forward<F>(f)
        );
        std::future<return_type> result = task->get_future();
        {
            std::lock_guard<std::mutex> lock(m_mutex);
            if (m_stop) throw std::runtime_error("enqueue on stopped ThreadPool");
            m_tasks.emplace([task](){ (*task)(); });
        }
        m_cv.notify_one();
        return result;
    }

    ~ThreadPool() {
        {
            std::lock_guard<std::mutex> lock(m_mutex);
            m_stop = true;
        }
        m_cv.notify_all();
        for (std::thread& worker : m_workers) {
            worker.join();
        }
    }

private:
    std::vector<std::thread> m_workers;
    std::queue<std::function<void()>> m_tasks;
    std::mutex m_mutex;
    std::condition_variable m_cv;
    bool m_stop;
};

有了线程池，我们的 Subject 就可以改成异步通知了。

template<typename... Args>
class AsyncSubject {
public:
    using Observer = std::function<void(Args...)>;

    explicit AsyncSubject(size_t threadCount = 4)
        : m_pool(std::make_shared<ThreadPool>(threadCount)) {}

    void attach(Observer obs) {
        std::lock_guard<std::mutex> lock(m_mutex);
        m_observers.push_back(std::move(obs));
    }

    void notify(Args... args) {
        std::vector<Observer> snapshot;
        {
            std::lock_guard<std::mutex> lock(m_mutex);
            snapshot = m_observers;  // 拷贝一份，避免迭代时被修改
        }

        for (auto& obs : snapshot) {
            // 每个观察者异步执行
            m_pool->enqueue([obs, args...]() {
                obs(args...);
            });
        }
    }

private:
    std::vector<Observer> m_observers;
    std::mutex m_mutex;
    std::shared_ptr<ThreadPool> m_pool;
};

核心变化：

  • notify 不再阻塞调用者，立即返回
  • 每个观察者的回调在线程池中独立执行
  • 通过拷贝观察者列表，避免锁竞争

避坑指南：我曾经踩过的坑

这个方案看起来挺完美，但实际用起来有几个坑。我曾经在一个实时数据推送系统里用过类似的设计，结果出了不少问题。

坑1：回调里加锁导致死锁

如果观察者的回调函数里又去 attach 或 detach 同一个 Subject，就会发生死锁。因为 notify 已经持有了 m_mutex，而 attach 也在等这个锁。

解决办法：notify 时先拷贝一份观察者列表，然后释放锁。这样回调里再操作 Subject 就不会死锁了。

坑2：观察者生命周期管理

如果观察者对象在回调执行期间被销毁了，程序会崩溃。我建议用 shared_ptr 管理观察者，或者在 detach 时确保没有正在执行的回调。

坑3：通知顺序不可控

异步通知意味着观察者收到通知的顺序是不确定的。如果你的业务逻辑依赖通知顺序，那就不能用纯异步方案。可以考虑用「有序线程池」或者给每个观察者分配独立的串行队列。

进阶：支持回调完成通知

有时候调用者想知道所有观察者都处理完了。我们可以让 notify 返回一个 future，等所有任务完成后再通知调用者。

std::shared_future<void> notify(Args... args) {
    std::vector<Observer> snapshot;
    {
        std::lock_guard<std::mutex> lock(m_mutex);
        snapshot = m_observers;
    }

    std::vector<std::future<void>> futures;
    for (auto& obs : snapshot) {
        auto fut = m_pool->enqueue([obs, args...]() {
            obs(args...);
        });
        futures.push_back(std::move(fut));
    }

    // 返回一个 future，等所有任务完成
    return std::async(std::launch::async, [futures = std::move(futures)]() {
        for (auto& f : futures) {
            f.wait();
        }
    }).share();
}

这样调用者可以这样用：

auto done = subject.notify(data);
done.wait();  // 等待所有观察者处理完毕

知识体系图

下面这张图总结了异步观察者模式的核心设计思路：

异步观察者模式架构 Subject attach / detach / notify ThreadPool 任务队列 + 工作线程 Observer列表 快照拷贝 Observer A Observer B Observer C ... 关键流程： 1. Subject.notify() 拷贝观察者列表快照，释放锁 2. 将每个观察者的回调封装为任务，提交到线程池 3. 线程池的工作线程从队列中取出任务并执行

面试中可能会追问的问题

问题 考察点 建议回答方向
如果观察者抛出异常怎么办？ 异常安全 在任务函数内 try-catch，或者用 std::future 捕获异常
如何避免观察者回调执行时间过长？ 超时控制 给每个任务设置超时，超时后强制取消或记录日志
线程池的线程数怎么确定？ 性能调优 一般用 std::thread::hardware_concurrency()，IO密集型可以多一些
能不能用 C++20 的 std::execution 来简化？ 新标准了解程度 可以，但需要编译器支持，生产环境可能不成熟

我的建议：

面试时不要只背代码。面试官更想听你讲「为什么这么设计」以及「遇到过什么坑」。你可以说：「我一开始也是同步通知，后来发现某个观察者写日志太慢，整个系统都卡住了，才改成异步的。」这样比单纯背代码要有说服力得多。

好了，关于异步观察者模式就聊这么多。这个模式在实际项目中非常常见，尤其是事件驱动架构、消息推送系统、游戏引擎这些场景。掌握好它，你的架构设计能力会上一个台阶。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →