一个多线程安全的通用订阅发布框架的实现

在企业开发中经常需要用到观察者模式：一些对象（Observer）作为观察者，对目标对象（Subject）的某些变化感兴趣,当变化发生时，Observer 需要及时获得通知，以便迅速做出改变或执行某些操作。

如果把这套“订阅—通知”的逻辑直接散落在业务代码里，很快就会遇到三个麻烦：订阅列表由谁维护、观察者销毁后留下的悬垂指针怎么处理、回调执行过程中又去注销自己导致死锁怎么办。本文实现一个通用的线程安全订阅通知框架 Notifier，并给出一个可编译运行的完整示例，说明如何在业务代码中低成本复用它。

一、设计目标

动手前先明确四条约束，后文的每个设计细节都是为它们服务的：

线程安全：注册、注销、广播可能发生在任意线程，所有共享数据都要用互斥锁保护。
生命周期安全：事件中心不“拥有”观察者。观察者销毁后自动脱离，既没有悬垂指针，也不会因为事件中心持有 shared_ptr 而导致对象永远无法释放。
低侵入：被观察者只需继承模板并在事件发生时调用一次广播。观察者只需实现自己的回调接口，不依赖被观察者的具体类型。
可复用：失效清理、加锁快照、参数转发这些容易写错的“脏活”只写一遍。

二、核心实现：Notifier 模板

完整的头文件实现如下，后面再逐点拆解关键设计。

//Notifier.h
#pragma once
#include <algorithm>
#include <functional>
#include <map>
#include <mutex>
#include <string>
#include <vector>
#include <memory>


template <typename CallbackT>
class Notifier
{
public:
    using Callback = CallbackT;
    using CallbackWeakPtr = std::weak_ptr<CallbackT>;
    using CallbackPtr = std::shared_ptr<CallbackT>;
    using CallbackList = std::vector<CallbackWeakPtr>;

    virtual ~Notifier() = default;

    // 生命周期安全：观察者以 shared_ptr 注册，但内部只保存 weak_ptr
    // 事件中心不“拥有”观察者，观察者销毁后自动从列表脱离，
    // 既避免悬垂指针，也避免“事件中心持有 shared_ptr 导致对象永远无法释放”的经典泄漏
    void registerCallback(CallbackPtr callback)
    {
        if (callback == nullptr)
            return;

        //注册/注销/广播可能发生在任意线程，所有共享数据都用互斥锁保护
        std::lock_guard<std::mutex> lock(mDataLock);
        //检查一下，确保之前没加入过的callback对象再插入vector，不然会导致重复通知
        if (std::find_if(mCallbacks.begin(), mCallbacks.end(), [callback](CallbackWeakPtr mWeakCallback) {
            if (auto mCallback = mWeakCallback.lock())
                return callback == mCallback;
            return false;
            }) == mCallbacks.end())
        {
            mCallbacks.push_back(callback);
        }
    }

    void unRegisterCallback(CallbackPtr callback)
    {
        if (callback == nullptr)
            return;

        std::lock_guard<std::mutex> lock(mDataLock);
        auto iter = std::find_if(mCallbacks.begin(), mCallbacks.end(), [callback](CallbackWeakPtr mWeakCallback) {
            if (auto mCallback = mWeakCallback.lock())
            {
                return callback == mCallback;
            }
            return false;
        });

        if (iter != mCallbacks.end())
        {
            mCallbacks.erase(iter);
        }
    }

    //当前的观察者注册数量（供调试/测试使用）
    size_t getRegisteredCallbacksCount()
    {
        std::lock_guard<std::mutex> lock(mDataLock);
        return mCallbacks.size();
    }

protected:
    CallbackList mCallbacks;
    std::mutex mDataLock;

    using Closure = std::function<void(std::shared_ptr<Callback>)>;

    // “只广播给某类观察者”的能力，但又不应直接操作 mCallbacks（会破坏封装/线程安全）。
    // 这里通过 shouldFire 谓词把过滤能力下放给子类，基类仍统一负责
    // 清理失效项、加锁快照、逐个 lock() 保证存活——这些“脏活”只写一遍。
    void potentiallyFireNotification(Closure closure, std::function<bool(CallbackWeakPtr)> shouldFire = [](CallbackWeakPtr) -> bool { return true; })
    {
        CallbackList liveCallbacks = removeExpiredCallbacks();

        for (auto callback : liveCallbacks)
        {
            if (auto lockedCallback = callback.lock())
            {
                if (shouldFire(lockedCallback))
                {
                    closure(lockedCallback);
                }
            }
        }
    }

    template <typename T>
    const T& to_const_ref_or_copy(T& arg)
    {
        return arg;
    }

    template <typename T>
    T to_const_ref_or_copy(T&& arg)
    {
        return std::remove_reference_t<T>{ arg };
    }

    // 向所有存活观察者调用成员函数 func(args...)。
    //
    // 参数转发策略：
    // - 只有 1 个观察者：可以安全地完美转发（std::forward），支持 move-only 参数、零拷贝。
    // - 有多个观察者：退化为 to_const_ref_or_copy——
    //   左值传 const 引用、右值拷贝，避免第一个回调把右值“消费掉”导致后续回调数据被掏空。
    //   换句话说：宁可多一次拷贝，也要保证每个订阅者收到语义一致的参数。
    template <typename Func, typename... Args>
    void fireNotification(Func func, Args&&... args)
    {
        const auto liveCallbacks = removeExpiredCallbacks();
        const auto enablePerfectForwarding = liveCallbacks.size() == 1;

        for (const auto& callback : liveCallbacks)
        {
            if (auto lockedCallback = callback.lock())
            {
                if (enablePerfectForwarding)
                {
                    std::invoke(func, lockedCallback, std::forward<Args>(args)...);
                }
                else
                {
                    std::invoke(func, lockedCallback, to_const_ref_or_copy(std::forward<Args>(args))...);
                }
            }
        }
    }

    // 钩子：当某个观察者的 weak_ptr 失效、被清理时回调子类。
    // 子类若维护了“旁路数据结构”（如按 key 的关联表），可在此同步清理。
    // 默认什么都不做。注意：调用时 mDataLock 已被持有，实现中不要再加锁。
    virtual void callbackErased(CallbackWeakPtr)
    {
    }

private:
    // 剔除已失效的 weak_ptr，并返回存活列表的“快照”。
    //
    // 为什么要返回快照而不是边遍历边广播：
    // 1. 死锁：回调内部常常会调用 registerCallback/unRegisterCallback，
    //    如果遍历期间还持有 mDataLock，同一线程会直接死锁（std::mutex 不可重入）
    // 2. 迭代器失效：回调可能在执行中注销自己/他人，修改正在遍历的容器会导致 UB
    // 因此这里的模式固定为：加锁 -> 拷贝存活项 + 清理失效项 -> 解锁 -> 遍历快照执行回调
    CallbackList removeExpiredCallbacks()
    {
        CallbackList liveCallbacks;
        {
            std::lock_guard<std::mutex> lock(mDataLock);

            if (mCallbacks.empty())
            {
                return {};
            }

            //提前reserve提高vector性能，防止频繁扩容
            liveCallbacks.reserve(mCallbacks.size());

            for (auto callbackIter = mCallbacks.begin(); callbackIter != mCallbacks.end(); ++callbackIter)
            {
                auto callback = *callbackIter;
                if (!callback.expired())
                {
                    liveCallbacks.push_back(callback);
                }
                else
                {
                    //源对象观察者已经销毁了
                    callbackErased(callback);
                }
            }

            if (mCallbacks.size() != liveCallbacks.size())
            {
                mCallbacks = liveCallbacks;
            }
        }

        return liveCallbacks;
    }
};

接口一览

成员	访问级别	作用
`registerCallback()`	public	注册观察者（传入 `shared_ptr`），内部自动去重
`unRegisterCallback()`	public	主动注销观察者
`getRegisteredCallbacksCount()`	public	当前注册数量，便于调试与测试
`fireNotification()`	protected	向所有存活观察者广播事件
`potentiallyFireNotification()`	protected	带 `shouldFire` 谓词的条件广播
`callbackErased()`	protected virtual	观察者失效时的清理钩子，供子类同步旁路数据
`removeExpiredCallbacks()`	private	清理失效 `weak_ptr` 并返回存活快照

三个关键设计

1. 只保存 weak_ptr，事件中心不拥有观察者。
注册时接收 shared_ptr，存储时转成 weak_ptr。观察者的生死由其创建者决定，事件中心只是一个“通讯录”。通讯录里某一项失效了，广播时顺手划掉即可，不会延长任何对象的生命周期。

2. 先加锁快照，再解锁广播。
removeExpiredCallbacks() 在锁内完成三件事：拷贝存活项、清理失效项、回写容器。锁释放之后才开始遍历快照执行回调。这样即使回调内部又去注册、注销自己或他人，也不会出现 std::mutex 重入死锁，更不会踩到迭代器失效的未定义行为。

3. 参数转发按观察者数量切换策略。
只有一个观察者时可以放心完美转发，支持 move-only 参数、零拷贝。有多个观察者时退化为 to_const_ref_or_copy——左值传 const 引用、右值拷贝一份，避免第一个回调把右值“消费”掉之后，后续观察者拿到被掏空的数据。

扩展点：条件广播与清理钩子

potentiallyFireNotification() 把“通知谁”的过滤逻辑以 shouldFire 谓词的形式下放给子类，而快照、清理、加锁、lock() 保活这些共性逻辑仍由基类统一负责。callbackErased() 则给维护了旁路数据结构的子类（例如按 key 建索引的通知器）留了一个同步清理的时机，调用时锁已被持有，实现中不要再加锁。

三、使用示例：一匹马

假设有一匹马（被观察者），我们对它的“跳”“吃”“睡”三个动作感兴趣。观察者实现回调接口 IHorseCallback，在这三类事件发生时执行打印动作。

3.1 回调接口与被观察者

//Horse.h
#pragma once
#include <iostream>
#include <string>
#include "Notifier.h"

// 观察者回调接口：实现它即可订阅马的一举一动
class IHorseCallback
{
public:
    virtual ~IHorseCallback() = default;

    virtual void onJump(int heightCm) = 0;
    virtual void onEat(const std::string& food) = 0;
    virtual void onSleep(int minutes) = 0;

    // 供 potentiallyFireNotification 过滤使用的兴趣标记，默认不关心
    virtual bool careAboutRace() const { return false; }
    virtual void onRaceStart() {}
};

// 被观察者：马。
// 它只负责两件事：1) 表达“我发生了什么” 2) 把事件广播出去。
// 至于“谁在听、听了做什么”，Horse 完全不需要知道，也不持有观察者的 shared_ptr。
class Horse : public Notifier<IHorseCallback>
{
public:
    void jump(int heightCm)
    {
        std::cout << "[Horse] jumped " << heightCm << " cm" << std::endl;
        fireNotification(&IHorseCallback::onJump, heightCm);
    }

    void eat(const std::string& food)
    {
        std::cout << "[Horse] eating " << food << std::endl;
        fireNotification(&IHorseCallback::onEat, food);
    }

    void sleep(int minutes)
    {
        std::cout << "[Horse] sleeping " << minutes << " min" << std::endl;
        fireNotification(&IHorseCallback::onSleep, minutes);
    }

    // 只有声明“关心比赛”的观察者才会收到这次通知，见第四节
    void raceStart()
    {
        potentiallyFireNotification(
            [](CallbackPtr observer) { observer->onRaceStart(); },
            [](CallbackWeakPtr weak) {
                auto observer = weak.lock();
                return observer != nullptr && observer->careAboutRace();
            });
    }
};

这里的 fireNotification(&IHorseCallback::onJump, heightCm) 值得单独说明：第一个参数是回调接口的成员函数指针，它与 std::shared_ptr<IHorseCallback> 一起构成 std::invoke 的调用对象，因此子类只需要描述“发生了什么”，不必自己写遍历循环。

3.2 观察者

//HorseWatcher.h
#pragma once
#include <iostream>
#include <string>
#include <utility>
#include "Horse.h"

// 观察者：实现回调接口，收到事件后打印一行日志
class HorseWatcher : public IHorseCallback
{
public:
    explicit HorseWatcher(std::string name) : mName(std::move(name)) {}

    void onJump(int heightCm) override
    {
        std::cout << "  [" << mName << "] 打印：马跳了，高度 " << heightCm << " cm" << std::endl;
    }

    void onEat(const std::string& food) override
    {
        std::cout << "  [" << mName << "] 打印：马吃了 " << food << std::endl;
    }

    void onSleep(int minutes) override
    {
        std::cout << "  [" << mName << "] 打印：马睡了 " << minutes << " 分钟" << std::endl;
    }

private:
    std::string mName;
};

3.3 组装与运行

源文件保存为 UTF-8，用 g++ -std=c++17 main.cpp -o demo 即可编译运行。

//main.cpp
#include <iostream>
#include <memory>
#include "Horse.h"
#include "HorseWatcher.h"

int main()
{
    Horse horse;

    auto watcher1 = std::make_shared<HorseWatcher>("Watcher-1");
    auto watcher2 = std::make_shared<HorseWatcher>("Watcher-2");

    horse.registerCallback(watcher1);
    horse.registerCallback(watcher2);
    horse.registerCallback(watcher2);   // 重复注册会被去重，仍然是 2 个
    std::cout << "registered = " << horse.getRegisteredCallbacksCount() << std::endl << std::endl;

    horse.jump(120);
    horse.eat("apple");
    horse.sleep(30);

    // 主动注销：Watcher-2 不再收到任何通知
    horse.unRegisterCallback(watcher2);
    std::cout << std::endl << "after unregister watcher2" << std::endl;
    horse.jump(60);

    // 观察者销毁：内部只保存 weak_ptr，因此不会泄漏，也不需要手动注销
    watcher1.reset();
    std::cout << std::endl << "after watcher1 destroyed" << std::endl;
    horse.jump(30);                     // 只剩 Horse 自己的日志，没有观察者打印
    std::cout << "registered = " << horse.getRegisteredCallbacksCount() << std::endl;

    return 0;
}

运行结果：

registered = 2

[Horse] jumped 120 cm
  [Watcher-1] 打印：马跳了，高度 120 cm
  [Watcher-2] 打印：马跳了，高度 120 cm
[Horse] eating apple
  [Watcher-1] 打印：马吃了 apple
  [Watcher-2] 打印：马吃了 apple
[Horse] sleeping 30 min
  [Watcher-1] 打印：马睡了 30 分钟
  [Watcher-2] 打印：马睡了 30 分钟

after unregister watcher2
[Horse] jumped 60 cm
  [Watcher-1] 打印：马跳了，高度 60 cm

after watcher1 destroyed
[Horse] jumped 30 cm
registered = 0

最后一行的 registered = 0 值得注意：watcher1.reset() 之后，mCallbacks 里那个 weak_ptr 已经失效，但它是在下一次广播（removeExpiredCallbacks）时才被顺手清理掉的，Horse 完全不需要感知观察者的生命周期。

3.4 示例要点

被观察者极薄：Horse 里没有锁、没有容器、没有生命周期管理，新增事件就是新增一个成员函数。
接口隔离：观察者只依赖 IHorseCallback，既不需要继承 Notifier，也不需要知道 Horse 的存在，可以自由复用、单独测试。
所有权清晰：Horse 不拥有观察者。观察者无论是 UI 面板、缓存对象还是临时对象都无所谓，谁创建谁持有，销毁即自动脱离。
线程安全：多个线程同时调用 jump()/eat()/sleep() 与 registerCallback()/unRegisterCallback() 都是安全的。回调执行期间不持锁，回调内部再注册或注销也不会死锁。

四、进阶：只通知部分观察者

上面的 raceStart() 用到了基类的 potentiallyFireNotification()：在 Horse 一侧声明“这次通知只发给关心比赛的观察者”，过滤谓词由子类提供，而快照、清理失效项、加锁、逐个 lock() 这些容易出错的活依然由基类统一完成。例如再写一个只关心比赛、不关心吃喝睡的观察者：

class RaceCommentator : public IHorseCallback
{
public:
    void onJump(int) override {}
    void onEat(const std::string&) override {}
    void onSleep(int) override {}

    bool careAboutRace() const override { return true; }

    void onRaceStart() override
    {
        std::cout << "  [Commentator] 打印：比赛开始，马已就位" << std::endl;
    }
};

把它和 HorseWatcher 一起注册后调用 horse.raceStart()，只有 RaceCommentator 会打印。同一个事件中心，既支持“广播给所有人”，也支持“广播给某一类人”。

五、小结

Notifier 把订阅通知中最容易出错的部分——生命周期、并发、参数语义——一次性封装干净，业务侧只剩下“表达事件”和“处理事件”两件事。落地时记住三条约定即可：

观察者一律用 shared_ptr 持有，注册后不必手动注销，销毁即自动脱离。
广播只在事件发生的线程上同步执行，回调里不要做长时间的阻塞操作。确需耗时处理时，转投到自己的线程池。
需要按类型、按兴趣定向通知时，优先扩展 shouldFire 谓词，而不是在子类里直接操作 mCallbacks。