/** * 线程安全版的event,可以用于多线程使用 * 使用方法: * 1. 创建ParallelEvent对象 * ParallelEvent ev{}; * 2. 建立连接,并保留句柄 * auto connection = ev.Subscribe( ... ); * 3. 发射事件 * ev.EmitEvent( ... ); * 4. 取消订阅, 直接析构connection句柄即可 */ #ifndef ZCPPWIN_PARALLEL_EVENT_HPP #define ZCPPWIN_PARALLEL_EVENT_HPP #include #include #include #include #include #include #include #include template class ParallelEvent { public: using Callback = std::function; class Connection { // 返回给客户的句柄,析构会Disconnect public: Connection() = default; Connection(const Connection &) = delete; Connection &operator=(const Connection &) = delete; Connection(Connection &&other) noexcept { MoveFrom(std::move(other)); } Connection &operator=(Connection &&other) noexcept { if (this != &other) { Disconnect(); MoveFrom(std::move(other)); } return *this; } ~Connection() { Disconnect(); } void Disconnect() { if (owner_) { owner_->Unsubscribe(id_); owner_ = nullptr; id_ = 0; } } explicit operator bool() const noexcept { return owner_ != nullptr; } private: friend class ParallelEvent; Connection(ParallelEvent *owner, std::uint64_t id) : owner_(owner), id_(id) {} void MoveFrom(Connection &&other) noexcept { owner_ = other.owner_; id_ = other.id_; other.owner_ = nullptr; other.id_ = 0; } private: ParallelEvent *owner_ = nullptr; std::uint64_t id_ = 0; }; ParallelEvent() = default; ParallelEvent(const ParallelEvent &) = delete; ParallelEvent &operator=(const ParallelEvent &) = delete; // 订阅 // 例如:auto guard = shared_from_this(); event.subscribe(guard, [...]{}); template Connection Subscribe(std::weak_ptr guard, Callback cb) { Handler h; h.id = nextId_++; h.guard = std::move(guard); h.cb = std::move(cb); // { std::lock_guard lk(mtx_); handlers_.emplace(h.id, std::move(h)); } return Connection(this, h.id); } Connection Subscribe(Callback cb) { Handler h; h.id = nextId_++; h.cb = std::move(cb); // { std::lock_guard lk(mtx_); handlers_.emplace(h.id, std::move(h)); } return Connection(this, h.id); } // 订阅 + 指定派发器(想跨线程就传一个 dispatcher) // dispatcher 接收一个 “要执行的函数”,你可以用 invokeMethod / 线程池等方式调度 template Connection Subscribe(std::weak_ptr guard, Callback cb, std::function)> dispatcher) { Handler h; h.id = nextId_++; h.guard = std::move(guard); h.cb = std::move(cb); h.dispatcher = std::move(dispatcher); // { std::lock_guard lk(mtx_); handlers_.emplace(h.id, std::move(h)); } return Connection(this, h.id); } // 发射事件(线程安全) void EmitEvent(Args... args) { // 1) 快照,避免回调里修改订阅造成死锁 std::vector snapshot; snapshot.reserve(64); // { std::lock_guard lk(mtx_); snapshot.reserve(handlers_.size()); for (auto &kv: handlers_) snapshot.push_back(kv.second); } // 2) 调用(不持锁) bool needCleanup = false; for (Handler &h: snapshot) { if (!IsAlive(h)) { needCleanup = true; continue; } // 如果绑定了 dispatcher,优先走 dispatcher if (h.dispatcher) { std::function fn = MakeInvokeFn(h, args...); h.dispatcher(std::move(fn)); continue; } // 非 QObject:直接调用 h.cb(args...); } // 3) 懒清理:有失效订阅时,移除之(避免 map 越积越多) if (needCleanup) CleanupExpired(); } // 便利别名 void operator()(Args... args) { EmitEvent(std::forward(args)...); } // 主动清理全部失效订阅 void CleanupExpired() { std::lock_guard lk(mtx_); for (auto it = handlers_.begin(); it != handlers_.end();) { if (!IsAlive(it->second)) it = handlers_.erase(it); else ++it; } } // 主动清空 void Clear() { std::lock_guard lk(mtx_); handlers_.clear(); } private: struct Handler { std::uint64_t id = 0; std::weak_ptr guard; // 生命周期守卫(可空) Callback cb; // 跨线程派发器(可选) std::function)> dispatcher; }; void Unsubscribe(std::uint64_t id) { std::lock_guard lk(mtx_); handlers_.erase(id); } static bool IsAlive(const Handler &h) { if (!h.guard.expired()) { return true; } // 如果 guard 过期,认为失效 // (允许“无 guard 订阅”,那就别用这个 overload) // 比较guard和nullptr的wp之间的控制块是否相同 if (h.guard.owner_before(std::weak_ptr{}) || std::weak_ptr{}.owner_before(h.guard)) { return !h.guard.expired(); } // h.guard 为空 weak_ptr:当作永远存活(用户自行持有 Connection 管理) return true; } template static std::function MakeInvokeFn(const Handler &h, A &&... a) { // 捕获回调 + 参数副本,确保异步安全 auto cb = h.cb; auto tup = std::make_tuple(std::forward(a)...); return [cb = std::move(cb), tup = std::move(tup)]() mutable { std::apply(cb, tup); }; } private: std::mutex mtx_; std::unordered_map handlers_; std::atomic nextId_{1}; }; #endif //ZCPPWIN_PARALLEL_EVENT_HPP