C++观察者模式与多线程事件通知实现（观察者.多线程.模式.事件.通知...）

答案：C++多线程观察者模式通过线程安全的观察者列表管理、异步事件分发、weak_ptr避免循环引用、事件队列与工作线程解耦通知过程，确保并发环境下的安全性与高性能。

在C++中实现观察者模式并结合多线程进行事件通知，本质上是为了构建一个解耦、响应迅速且可扩展的系统。当一个“被观察者”（Subject）发生状态变化时，它能够通知所有注册的“观察者”（Observer），而多线程的引入则让这个通知过程可以异步进行，避免阻塞主逻辑，并允许观察者在各自的线程中处理事件，这对于构建高性能、低延迟的并发应用至关重要。

解决方案

要实现一个健壮的C++多线程观察者模式，我们需要关注几个核心点：线程安全的观察者列表管理、异步事件分发机制、以及观察者生命周期的妥善处理。

首先，定义观察者和被观察者的接口。一个

IObserver

接口通常会有一个

update

方法，接收事件数据。

ISubject

接口则包含

attach

、

detach

和

notify

方法。

// 事件基类，用于传递不同类型的事件数据
struct Event {
    enum Type {
        Generic,
        DataChanged,
        StatusUpdate
        // ...更多事件类型
    };
    Type type = Generic;
    // 可以添加时间戳、源ID等通用信息
    virtual ~Event() = default;
};

// 观察者接口
class IObserver {
public:
    virtual ~IObserver() = default;
    virtual void update(const Event& event) = 0;
};

// 被观察者接口
class ISubject {
public:
    virtual ~ISubject() = default;
    virtual void attach(std::shared_ptr<IObserver> observer) = 0;
    virtual void detach(std::shared_ptr<IObserver> observer) = 0;
    virtual void notify(const Event& event) = 0;
};

接着，实现一个具体的

Subject

类。这里需要重点考虑线程安全。观察者列表（

std::vector<std::weak_ptr<IObserver>>

是个不错的选择，避免循环引用）的增删改查必须由互斥锁（

std::mutex

）保护。

#include <vector>
#include <memory>
#include <mutex>
#include <algorithm> // for std::remove_if
#include <iostream>  // for example output

// 前面定义的Event和IObserver/ISubject接口

class ConcreteSubject : public ISubject {
public:
    void attach(std::shared_ptr<IObserver> observer) override {
        std::lock_guard<std::mutex> lock(mtx_);
        // 避免重复添加
        for (const auto& w_obs : observers_) {
            if (auto s_obs = w_obs.lock(); s_obs == observer) {
                return;
            }
        }
        observers_.push_back(observer);
    }

    void detach(std::shared_ptr<IObserver> observer) override {
        std::lock_guard<std::mutex> lock(mtx_);
        observers_.erase(
            std::remove_if(observers_.begin(), observers_.end(),
                [&](const std::weak_ptr<IObserver>& w_obs) {
                    return w_obs.lock() == observer;
                }),
            observers_.end());
    }

    void notify(const Event& event) override {
        // 这里的通知可以是同步的，也可以是异步的
        // 在多线程环境中，通常会选择异步通知
        // 对于同步通知，需要注意锁的粒度和潜在的死锁风险
        std::vector<std::shared_ptr<IObserver>> active_observers;
        {
            std::lock_guard<std::mutex> lock(mtx_);
            // 复制一份活跃的观察者列表，避免在通知过程中修改列表
            // 同时清理已失效的weak_ptr
            observers_.erase(
                std::remove_if(observers_.begin(), observers_.end(),
                    [&](const std::weak_ptr<IObserver>& w_obs) {
                        if (auto s_obs = w_obs.lock()) {
                            active_observers.push_back(s_obs);
                            return false; // 仍然有效
                        }
                        return true; // 已失效，移除
                    }),
                observers_.end());
        }

        // 异步通知的实现通常会将事件放入一个队列，由专门的线程处理
        // 这里只是同步通知的示例，实际多线程会更复杂
        for (const auto& observer : active_observers) {
            // 在多线程场景下，这里不应该直接调用update，而是将事件推送到一个队列
            // observer->update(event); // 同步调用会阻塞
            // 假设我们有一个异步事件分发器
            EventDispatcher::getInstance().dispatchEvent(observer, event);
        }
    }

private:
    std::vector<std::weak_ptr<IObserver>> observers_;
    std::mutex mtx_; // 保护observers_列表
};

异步通知是多线程环境的关键。我们可以引入一个全局的

EventDispatcher

单例，它内部维护一个线程安全的事件队列和一组工作线程。当

Subject

调用

notify

时，它不是直接调用

Observer::update

，而是将事件和目标观察者（或其ID）打包，推送到

EventDispatcher

的队列中。工作线程从队列中取出事件，并在自己的线程中调用观察者的

update

方法。

#include <queue>
#include <thread>
#include <condition_variable>
#include <atomic>

// 前面定义的Event和IObserver接口

class EventDispatcher {
public:
    static EventDispatcher& getInstance() {
        static EventDispatcher instance;
        return instance;
    }

    void dispatchEvent(std::shared_ptr<IObserver> observer, const Event& event) {
        {
            std::lock_guard<std::mutex> lock(queue_mtx_);
            event_queue_.push({observer, std::make_shared<Event>(event)}); // 复制事件，确保生命周期
        }
        cv_.notify_one(); // 通知一个工作线程有新事件
    }

    void start(int num_threads = 2) {
        if (running_.exchange(true)) return; // 避免重复启动
        for (int i = 0; i < num_threads; ++i) {
            worker_threads_.emplace_back(&EventDispatcher::worker_loop, this);
        }
    }

    void stop() {
        if (!running_.exchange(false)) return; // 避免重复停止
        cv_.notify_all(); // 唤醒所有等待的线程
        for (std::thread& t : worker_threads_) {
            if (t.joinable()) {
                t.join();
            }
        }
        worker_threads_.clear();
    }

private:
    struct QueuedEvent {
        std::shared_ptr<IObserver> observer;
        std::shared_ptr<Event> event_data; // 使用shared_ptr管理事件数据生命周期
    };

    EventDispatcher() = default;
    ~EventDispatcher() {
        stop(); // 确保在析构时停止所有线程
    }
    EventDispatcher(const EventDispatcher&) = delete;
    EventDispatcher& operator=(const EventDispatcher&) = delete;

    void worker_loop() {
        while (running_) {
            QueuedEvent q_event;
            {
                std::unique_lock<std::mutex> lock(queue_mtx_);
                cv_.wait(lock, [this]{ return !running_ || !event_queue_.empty(); });

                if (!running_ && event_queue_.empty()) {
                    break; // 停止信号且队列为空，退出循环
                }

                if (!event_queue_.empty()) {
                    q_event = event_queue_.front();
                    event_queue_.pop();
                } else {
                    continue; // 队列为空但未停止，继续等待
                }
            } // 解锁，允许其他线程添加事件

            if (q_event.observer) { // 检查观察者是否仍然存在
                q_event.observer->update(*q_event.event_data);
            }
        }
    }

    std::queue<QueuedEvent> event_queue_;
    std::mutex queue_mtx_;
    std::condition_variable cv_;
    std::vector<std::thread> worker_threads_;
    std::atomic<bool> running_ = false;
};

C++观察者模式在多线程环境下如何确保线程安全？

确保C++观察者模式在多线程环境下的线程安全，在我看来，主要涉及到对共享资源的保护和事件通知机制的选择。这可不是简单地加几个锁就能万事大吉的，需要一套组合拳。

首先，最直观也是最关键的一点，就是保护观察者列表。

Subject

内部维护的

std::vector<std::weak_ptr<IObserver>>

这个列表是所有线程共享的资源，当有线程尝试

attach

或

detach

观察者，或者

notify

时遍历这个列表，都可能引发竞态条件。所以，

std::mutex

是必不可少的。每次对

observers_

列表进行读写操作时，都应该用

std::lock_guard

或

std::unique_lock

进行加锁。这能保证同一时间只有一个线程在修改或遍历列表，避免了迭代器失效、数据不一致等问题。

其次，妥善处理观察者的生命周期。这是个老生常谈但又容易出错的地方。如果

Subject

持有

std::shared_ptr<IObserver>

，而

IObserver

又反过来持有

std::shared_ptr<ISubject>

，就会形成循环引用，导致内存泄漏。

std::weak_ptr

在这里就显得尤为重要。

Subject

持有

std::weak_ptr<IObserver>

，这样即使观察者被销毁了，

Subject

也能安全地检测到（通过

weak_ptr::lock()

返回

nullptr

），并将其从列表中移除，避免了访问悬空指针的风险。这种机制也意味着观察者可以独立于被观察者而被销毁，解耦性更好。

再者，事件通知的同步与异步选择。在多线程环境下，直接在

notify

方法中同步调用所有观察者的

update

方法是非常危险的。这不仅可能阻塞

Subject

所在的线程，如果某个观察者的

update

方法耗时过长，甚至可能导致整个系统响应迟钝。更糟糕的是，如果观察者内部也持有锁并尝试获取

Subject

的锁，就可能导致死锁。因此，异步通知几乎是多线程观察者模式的黄金标准。将事件放入一个线程安全的队列，由专门的事件分发线程或线程池来处理，能彻底解耦事件产生和事件处理的过程，极大地提升系统的并发性和鲁棒性。

最后，事件数据本身的生命周期管理。当事件数据从

Subject

传递给

Observer

时，谁来负责它的创建和销毁？如果事件数据是简单的值类型，按值传递或

const&

传递通常没问题。但对于复杂或大型事件数据，使用

std::shared_ptr<const Event>

来传递是个好办法。这样，事件数据在所有需要它的观察者处理完毕之前都不会被销毁，并且可以避免不必要的拷贝。

在我看来，线程安全不仅仅是技术细节，更是一种设计哲学，它要求我们对系统中所有共享状态和并发操作有清晰的认识和预判。

PIA

全面的AI聚合平台，一站式访问所有顶级AI模型

226 查看详情 异步事件通知机制在C++多线程观察者模式中的优势与实现细节？

异步事件通知机制在C++多线程观察者模式中，可以说是一剂“灵丹妙药”，它带来的优势是显而易见的，并且实现起来也有其精妙之处。

核心优势：

1. 解耦与非阻塞： 这是最显著的优点。当

   Subject

   发出事件时，它不需要等待所有

   Observer

   处理完毕。它只需将事件投递到队列中，然后立即返回，继续执行自己的核心业务逻辑。这样就避免了

   Subject

   线程被任何一个缓慢或复杂的

   Observer

   阻塞，极大地提升了系统的响应性和吞吐量。在我看来，一个高性能系统，其核心组件绝不应该被非核心逻辑所拖累。
2. 并发性提升： 事件处理可以由独立的线程池来完成。这意味着多个事件可以并行处理，或者同一个事件可以被多个观察者并行处理（如果观察者本身是线程安全的）。这比同步通知的串行处理效率高出不止一个数量级。
3. 容错性与稳定性： 异步机制可以有效隔离故障。如果一个观察者的

   update

   方法抛出异常或者陷入死循环，它只会影响到处理该事件的那个工作线程，而不会波及到

   Subject

   线程或其他观察者。这大大提高了系统的整体稳定性。
4. 负载均衡： 如果采用线程池来处理事件，当事件量激增时，线程池可以根据负载情况动态调整工作线程数量（如果实现得足够高级），或者至少通过队列来平滑处理峰值，避免系统崩溃。

实现细节：

异步事件通知的实现，通常围绕着一个线程安全的队列和一组工作线程（或一个单独的事件分发线程）展开，这本质上是一个生产者-消费者模型。

1. 线程安全队列：

   • 数据结构：

     std::queue

     是一个常见的选择，但它本身不是线程安全的。
   • 同步原语： 必须结合

     std::mutex

     来保护队列的入队（

     push

     ）和出队（

     pop

     ）操作，防止竞态条件。
   • 通知机制：

     std::condition_variable

     是队列为空时等待、有新事件时唤醒工作线程的关键。生产者（

     Subject

     ）在

     push

     后调用

     notify_one()

     或

     notify_all()

     ，消费者（工作线程）在

     pop

     前调用

     wait()

     。
   • 存储内容： 队列中存储的应该是事件本身以及目标观察者的信息。通常，我们会存储

     std::shared_ptr<IObserver>

     和

     std::shared_ptr<Event>

     。使用

     shared_ptr

     是为了确保在事件处理完成之前，观察者对象和事件数据都不会被销毁。

2. 事件分发器（Dispatcher）与工作线程：

   • Dispatcher角色： 可以是一个单例类，它负责管理事件队列和工作线程。它提供一个公共接口（如

     dispatchEvent

     ）供

     Subject

     调用。
   • 工作线程： 每个工作线程都运行一个无限循环（直到收到停止信号）。在循环中，它会尝试从队列中取出事件。如果队列为空，它会通过

     std::condition_variable::wait

     进入休眠状态，直到有新事件到来。
   • 事件处理： 取出事件后，工作线程会安全地调用相应观察者的

     update

     方法。这里需要再次检查

     shared_ptr<IObserver>

     是否仍然有效（即观察者是否已被销毁），以避免访问无效内存。
   • 优雅关闭： 这是个容易被忽视但非常重要的一点。当系统需要关闭时，必须有一种机制能够通知所有工作线程安全地退出循环，处理完队列中剩余的事件，然后

     join

     它们。这通常通过一个

     std::atomic<bool>

     标志位和

     std::condition_variable::notify_all()

     来实现。当

     running_

     标志设为

     false

     时，工作线程会被唤醒，检查标志，然后退出循环。

坦白说，实现一个高效且健壮的异步事件分发器，需要对C++并发编程有深入的理解，尤其是对生命周期管理、异常安全和锁的细粒度控制。

如何处理C++观察者模式中复杂的事件数据与回调函数？

在C++观察者模式中，随着系统复杂度的提升，事件数据不再是简单的整型或字符串，回调函数也不再是单一的

update

方法。处理这些复杂性，在我看来，是模式能够适应不同场景的关键。

处理复杂的事件数据：

1. 事件基类与派生类（多态）： 这是最经典也是最常用的方法。定义一个抽象的

   Event

   基类（通常包含一个虚析构函数和一些通用字段，比如事件类型枚举）。然后为每种具体的事件类型派生出子类，例如

   MouseClickEvent

   、

   KeyboardEvent

   、

   NetworkDataReceivedEvent

   等。

   struct MouseClickEvent : public Event {
       MouseClickEvent(int x, int y) : x_pos(x), y_pos(

以上就是C++观察者模式与多线程事件通知实现的详细内容，更多请关注知识资源分享宝库其它相关文章！

相关标签： go ai c++ ios 并发编程 red 有锁 多态 子类 析构函数 整型 const 回调函数 字符串 bool 循环 指针 数据结构 接口 值类型 Event 线程 多线程 值传递 空指针 并发 对象 事件 异步 负载均衡 大家都在看： C++如何使用模板实现迭代器类 C++如何处理复合对象中的嵌套元素 C++如何使用const修饰变量 C++内存模型与编译器优化理解 C++如何使用ofstream和ifstream组合操作文件