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

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;
};

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>

update