如何高效实现无昂贵拷贝的线程安全观察者设计模式？

你的原实现存在两个核心问题：

• 通知时持有锁调用update()，会拉长锁的持有时间，不仅会阻塞其他添加/删除观察者的操作，还可能引发死锁（比如update()内部又调用addObserver()或removeObserver()）。
• 如果为了避免持锁回调而每次通知都拷贝观察者列表，高频率通知场景下拷贝开销会非常大。

最优方案：基于写时复制（Copy-On-Write）的原子指针实现 #

针对观察者模式读多写少（通知操作远多于添加/删除）的特性，用「共享指针+原子指针+写时复制」的方案可以完美解决这两个问题：

• 读操作（通知）无锁，直接读取当前的观察者列表快照，避免锁竞争。
• 写操作（添加/删除）仅在修改时拷贝一次列表，而非每次通知都拷贝，大幅降低开销。
• 彻底避免持锁调用回调的风险。

完整代码实现

#include <mutex>
#include <vector>
#include <memory>
#include <atomic>

// 先定义Observer基类
class Observer {
public:
    virtual ~Observer() = default;
    virtual void update() = 0;
};

class Subject {
public:
    Subject() 
        : observers_(std::make_shared<std::vector<Observer*>>()) {}

    void addObserver(Observer* observer) {
        // 先拷贝当前列表（写时复制）
        auto current = observers_.load(std::memory_order_acquire);
        auto new_observers = std::make_shared<std::vector<Observer*>>(*current);
        new_observers->push_back(observer);
        // 原子替换旧列表，保证线程可见性
        observers_.store(new_observers, std::memory_order_release);
    }

    void removeObserver(Observer* observer) {
        auto current = observers_.load(std::memory_order_acquire);
        auto new_observers = std::make_shared<std::vector<Observer*>>();
        // 过滤掉要删除的观察者
        for (auto obs : *current) {
            if (obs != observer) {
                new_observers->push_back(obs);
            }
        }
        observers_.store(new_observers, std::memory_order_release);
    }

    void notifyObservers() {
        // 原子加载当前列表的快照，无锁操作
        auto current_observers = observers_.load(std::memory_order_acquire);
        // 遍历快照调用update，此时即使列表被修改也不影响当前遍历
        for (auto observer : *current_observers) {
            observer->update();
        }
    }

private:
    // 用原子指针存储共享的观察者列表
    std::atomic<std::shared_ptr<std::vector<Observer*>>> observers_;
};

关键细节解释 #

1. 原子指针与内存顺序：
   • std::atomic<std::shared_ptr<>>保证列表指针的读写是原子操作，memory_order_acquire和memory_order_release确保修改后的列表能被其他线程立刻看到，避免内存可见性问题。
2. 写时复制逻辑：
   • 添加/删除观察者时，先拷贝当前的列表到新的shared_ptr，修改新列表后再原子替换旧指针。这样写操作只会触发一次拷贝，而读操作完全无锁。
3. 通知操作无锁：
   • 通知时只需要加载当前的列表快照，遍历快照调用update()，此时即使有其他线程修改列表，也只会修改新的拷贝，不会影响正在遍历的旧列表，彻底避免了持锁回调的问题。

额外优化建议 #

• 观察者生命周期管理：如果观察者可能被提前销毁，建议将std::vector<Observer*>改为std::vector<std::weak_ptr<Observer>>，遍历的时候先lock成shared_ptr，避免悬空指针：

  void notifyObservers() {
      auto current_observers = observers_.load(std::memory_order_acquire);
      for (auto& weak_obs : *current_observers) {
          if (auto obs = weak_obs.lock()) {
              obs->update();
          }
      }
  }
  

• 批量修改优化：如果需要频繁添加/删除观察者，可以加一个临时互斥锁，把连续的写操作合并，减少拷贝次数。

内容的提问来源于stack exchange，提问作者Peter