1. 项目概述：为什么我们需要在C++中实现PubSub？

如果你正在处理一个需要多个组件相互通信的C++项目，尤其是当这些组件可能分布在不同的进程甚至不同的机器上时，你很快就会遇到一个经典难题：如何高效、解耦地传递消息？直接函数调用太紧密，轮询又太低效。这时，发布-订阅（Publish-Subscribe，简称PubSub）模型就成了一个非常自然的选择。

简单来说，PubSub模型定义了一种消息传递范式：发布者（Publisher）将消息发送到特定的主题（Topic），而不需要知道谁将接收它；订阅者（Subscriber）则表达对一个或多个主题的兴趣，并接收所有发布到这些主题的消息。两者通过一个称为“代理”（Broker）或“事件总线”（Event Bus）的中介完全解耦。在C++中实现它，意味着我们能在高性能的本地系统内，构建起类似现代分布式消息队列（如Kafka、RabbitMQ）的通信骨架，这对于游戏引擎的事件系统、微服务间的进程内通信、插件架构的数据流控制等场景至关重要。

最近的热搜词，像“C++多线程”、“C++ map”、“vscode配置c++”等，恰恰反映了开发者们正深入C++的实践领域，从环境搭建到核心数据结构，再到并发编程。实现一个PubSub模型，正是将这些知识点串联起来的绝佳实践：你需要用到std::map或std::unordered_map来管理主题与订阅者的映射，需要std::function和std::bind来处理回调，更需要谨慎地使用std::thread、std::mutex和std::condition_variable来保证线程安全。这不仅仅是一个功能实现，更是一次对现代C++核心特性的综合演练。

2. 核心设计思路与架构选型

在动手写代码之前，我们必须先想清楚几个关键问题：这个PubSub模型是同步调用还是异步派发？主题匹配是精确匹配还是支持通配符？消息的生命周期如何管理？订阅者回调的执行上下文是什么？这些设计决策将直接影响到最终实现的复杂度、性能和适用场景。

2.1 同步 vs. 异步：消息派发的核心抉择

这是第一个需要权衡的点。同步发布意味着当publish函数被调用时，它会立即、依次地调用所有订阅者的回调函数。这种方式实现简单，逻辑直观，发布者能立刻知道消息是否被成功处理。但其致命缺点是会阻塞发布者线程，如果某个订阅者的回调函数执行缓慢或陷入死循环，整个发布流程甚至系统都会被卡住。

// 一个简单的同步发布伪代码示意 void publish(const std::string& topic, const Message& msg) { std::lock_guard<std::mutex> lock(mutex_); auto it = subscribers_.find(topic); if (it != subscribers_.end()) { for (auto& callback : it->second) { callback(msg); // 直接在此线程同步调用 } } }

异步发布则将消息放入一个队列，由后台的工作线程（或线程池）从队列中取出并执行回调。发布者函数在将消息入队后便可立即返回，实现了非阻塞。这对于需要高响应性的系统（如UI主线程、网络IO线程）至关重要。但异步引入了队列管理、线程安全、消息顺序保证以及更复杂的错误处理（回调异常发生在另一个线程）等问题。

我个人的经验是，对于绝大多数应用场景，异步模型是更优的选择。它更好地体现了PubSub的解耦本质——发布者只负责“通知”，而不关心“处理”。我们可以结合std::queue或std::deque作为消息队列，使用std::condition_variable来通知工作线程有新消息到达。

2.2 主题匹配策略：从简单到灵活

最初级的实现是精确字符串匹配。订阅者订阅“sensor.temperature”，那么只有发布到完全相同的主题的消息才会被收到。这实现起来最简单，一个std::unordered_map<std::string, std::vector<Callback>>就够了。

但在复杂系统中，我们常常需要更灵活的匹配。例如，订阅者可能想接收所有传感器数据（sensor.*），或者所有以error.开头的日志。这就引入了通配符匹配，最常见的是*（匹配单层）和#（匹配多层）。例如，MQTT协议就采用了这种模式。实现通配符匹配需要将主题字符串按分隔符（如.）分割，并进行树形（Trie）或规则匹配，这会增加一些复杂度。

对于第一个版本的实现，我建议从精确匹配开始。它足以解决80%的问题，并且性能最优。当你的系统确实需要更灵活的消息路由时，再考虑引入通配符。你可以设计一个TopicMatcher接口，初期实现一个ExactMatcher，后期再轻松替换或增加一个WildcardMatcher。

2.3 关键数据结构设计

一个线程安全的PubSub核心通常围绕以下几个数据结构展开：

1. 订阅关系表：这是核心映射。由于我们选择精确匹配起步，可以使用std::unordered_map<std::string, std::vector<Subscription>>。键（Key）是主题字符串，值（Value）是该主题下所有订阅者的集合。这里我用了Subscription而不仅仅是std::function，因为一个订阅实体可能还需要包含订阅ID（用于取消订阅）、订阅者弱引用（防止回调对象已销毁）等元信息。
2. 消息队列（异步模式）：用于暂存待处理的消息。通常是一个std::queue<std::pair<std::string, Message>>或更复杂的结构。必须用互斥锁保护。
3. 线程同步原语：std::mutex用于保护共享数据（订阅表、消息队列），std::condition_variable用于在异步模式下通知工作线程。
4. 订阅者句柄：subscribe函数应该返回一个唯一的SubscriptionHandle（例如一个uint64_t的ID或一个std::shared_ptr<SubscriptionToken>）。这个句柄是后续unsubscribe操作的凭证。直接要求用户记住回调函数对象来取消订阅是非常不友好且容易出错的。

3. 分步实现一个线程安全的异步PubSub模型

下面，我们将一步步实现一个功能相对完整、线程安全的异步PubSub模型。这个实现将包含：异步消息队列、订阅/取消订阅、以及基本的生命周期管理。

3.1 定义核心类型与消息体

首先，我们定义一些基础类型。消息体（Message）不应该是一个简单的std::any，为了效率和类型安全，我们可以使用一个小的类型擦除容器，或者简单地定义一个包含主题和数据的结构体。这里为了通用性，我们使用std::any来承载任意数据，但实际项目中你可能需要更精细的设计（如std::variant或自定义消息基类）。

#include <any> #include <functional> #include <memory> #include <string> // 前向声明 class PubSub; // 订阅回调函数类型 using Callback = std::function<void(const std::string& topic, const std::any& message)>; // 订阅句柄，用于唯一标识一个订阅，以便安全取消 struct SubscriptionHandle { uint64_t id; // 唯一ID std::string topic; // 可以添加其他信息，如订阅者弱引用等 bool operator==(const SubscriptionHandle& other) const { return id == other.id; } }; // 一个订阅条目 struct Subscription { Callback callback; SubscriptionHandle handle; };

3.2 实现PubSub核心类

我们将主要功能封装在PubSub类中。它管理订阅表、消息队列和一个后台工作线程。

#include <atomic> #include <condition_variable> #include <deque> #include <mutex> #include <thread> #include <unordered_map> #include <vector> class PubSub { public: PubSub(); ~PubSub(); // 订阅主题，返回一个可用于取消订阅的句柄 SubscriptionHandle subscribe(const std::string& topic, Callback callback); // 取消订阅 void unsubscribe(const SubscriptionHandle& handle); // 异步发布消息 void publish(const std::string& topic, std::any message); // 停止后台线程（在析构时自动调用） void stop(); private: void workerThread(); // 后台工作线程函数 std::atomic<bool> running_{true}; // 控制工作线程生命周期 std::thread worker_; // 后台工作线程 // 保护以下所有共享数据 std::mutex mutex_; // 订阅表：主题 -> 订阅列表 std::unordered_map<std::string, std::vector<Subscription>> subscriptions_; // 消息队列：待处理的消息（主题 + 数据） std::deque<std::pair<std::string, std::any>> messageQueue_; // 用于通知工作线程有新消息或需要退出 std::condition_variable cv_; // 用于生成唯一的订阅ID std::atomic<uint64_t> nextSubscriptionId_{1}; };

3.3 构造函数、析构函数与线程管理

构造函数启动后台工作线程，析构函数负责安全地停止它。

PubSub::PubSub() { worker_ = std::thread(&PubSub::workerThread, this); } PubSub::~PubSub() { stop(); } void PubSub::stop() { if (running_.exchange(false)) { cv_.notify_all(); // 通知工作线程醒来检查退出条件 if (worker_.joinable()) { worker_.join(); } } } void PubSub::workerThread() { while (running_) { std::pair<std::string, std::any> message; { std::unique_lock<std::mutex> lock(mutex_); // 等待条件：线程被要求停止，或消息队列非空 cv_.wait(lock, [this]() { return !running_ || !messageQueue_.empty(); }); // 如果被唤醒是因为要停止且队列为空，则退出循环 if (!running_ && messageQueue_.empty()) { break; } // 取出队列头部的消息 if (!messageQueue_.empty()) { message = std::move(messageQueue_.front()); messageQueue_.pop_front(); } else { continue; // 理论上不会发生，为安全起见 } } // 释放锁，允许其他线程继续发布或订阅 // 在无锁状态下执行回调，避免死锁，也避免回调阻塞队列 const std::string& topic = message.first; const std::any& msgData = message.second; std::vector<Subscription> subscribersCopy; { std::lock_guard<std::mutex> lock(mutex_); auto it = subscriptions_.find(topic); if (it != subscriptions_.end()) { // 复制订阅者列表，防止回调中修改原列表导致迭代器失效 subscribersCopy = it->second; } } // 执行回调 for (const auto& sub : subscribersCopy) { if (sub.callback) { try { sub.callback(topic, msgData); } catch (const std::exception& e) { // 强烈建议：处理回调异常，至少记录日志 // std::cerr << "Callback error on topic \"" << topic << "\": " << e.what() << std::endl; } } } } }

关键点解析：

1. workerThread使用std::condition_variable::wait配合谓词，优雅地处理了“等待消息”和“等待停止”两种状态。
2. 在查找订阅者列表时，我们复制了一份（subscribersCopy）。这是至关重要的！因为订阅者的回调函数sub.callback是在锁外执行的。如果在回调函数内部，又调用了subscribe或unsubscribe来修改subscriptions_，就会导致死锁（我们的线程正持有mutex_，等待回调返回，而回调又试图获取mutex_）。复制列表避免了这个问题。
3. 回调被包裹在try-catch块中。这是必须的防御性编程。一个订阅者的回调崩溃不应该影响其他订阅者接收消息，也不应该导致整个工作线程崩溃。

3.4 实现订阅与取消订阅

订阅操作需要生成唯一ID，并将订阅信息存入对应主题的列表中。

SubscriptionHandle PubSub::subscribe(const std::string& topic, Callback callback) { std::lock_guard<std::mutex> lock(mutex_); uint64_t newId = nextSubscriptionId_.fetch_add(1, std::memory_order_relaxed); SubscriptionHandle handle{newId, topic}; Subscription sub{std::move(callback), handle}; subscriptions_[topic].push_back(std::move(sub)); return handle; }

取消订阅操作需要根据句柄找到对应的主题和订阅项并移除。这里有一个常见的陷阱：直接遍历vector并擦除元素会导致迭代器失效。更安全高效的做法是使用std::remove_if算法。

void PubSub::unsubscribe(const SubscriptionHandle& handle) { std::lock_guard<std::mutex> lock(mutex_); auto it = subscriptions_.find(handle.topic); if (it != subscriptions_.end()) { auto& subs = it->second; // 使用remove-erase惯用法 subs.erase( std::remove_if(subs.begin(), subs.end(), [&handle](const Subscription& sub) { return sub.handle == handle; }), subs.end() ); // 如果该主题的订阅列表为空，可以选择删除这个空条目以节省内存 if (subs.empty()) { subscriptions_.erase(it); } } }

3.5 实现异步发布

发布操作非常简单：将消息放入队列，然后通知工作线程。

void PubSub::publish(const std::string& topic, std::any message) { { std::lock_guard<std::mutex> lock(mutex_); messageQueue_.emplace_back(topic, std::move(message)); } // 锁在通知前释放，是良好的实践 cv_.notify_one(); // 通知一个等待的工作线程 }

4. 高级话题与性能优化

一个基础的PubSub模型已经搭建完成，但要用于生产环境，我们还需要考虑更多。

4.1 内存管理与对象生命周期

这是C++ PubSub实现中最容易出错的地方之一。问题核心是：订阅者对象（其成员函数被绑定为回调）可能比PubSub代理或主题的生命周期更短。

• 场景：一个对象Subscriber obj订阅了主题，随后obj被销毁。此时如果还有消息发布到该主题，回调将指向一个已销毁的对象，导致未定义行为（通常是崩溃）。
• 解决方案：
  1. 使用std::weak_ptr：这是最健壮的方式。要求订阅者必须由std::shared_ptr管理。在Subscription中存储std::weak_ptr<Subscriber>和一个指向成员函数的指针。在执行回调前，尝试将weak_ptr提升（lock()）为shared_ptr，如果提升失败，说明对象已销毁，则安全地忽略或移除该订阅。这需要订阅者类有固定的接口。
  2. 使用自定义的令牌（Token）生命周期：让subscribe返回一个std::shared_ptr<SubscriptionToken>，该Token持有回调。订阅者持有这个Token。当订阅者想取消订阅时，直接让Token析构（或调用其reset方法）。在Token的析构函数中，向PubSub发送取消请求。这利用了RAII思想，避免了手动调用unsubscribe。
  3. 在回调中使用弱引用检查：对于绑定成员函数的情况，可以在回调函数的第一行检查一个对象内的“存活标志”（例如一个std::atomic<bool>或std::shared_ptr<void>），如果对象已标记为无效，则直接返回。

在我的项目中，我通常采用方案1和方案2的结合。定义一个Subscriber基类，提供虚函数onMessage，内部使用weak_ptr管理。对于更通用的回调，则返回一个RAII风格的SubscriptionGuard对象，在其析构时自动取消订阅。

4.2 支持通配符主题匹配

如前所述，通配符极大地增加了灵活性。实现它意味着我们不能再用简单的unordered_map进行精确查找。我们需要一个主题树（Topic Trie）。

主题通常用斜杠/或点.分隔，例如home/living_room/temperature。我们可以将主题分割成段（["home", "living_room", "temperature"]）。树中的每个节点对应一段，节点包含该段下的订阅者列表，以及指向子节点（下一段）的映射。

对于通配符+（单层）和#（多层）：

• +：匹配当前层的任意一个段。在遍历树时，如果遇到+节点，需要同时搜索当前层的所有子节点。
• #：匹配当前层及以下所有层。它必须出现在主题末尾。在树中，#可以作为一个特殊的终止节点，当匹配到它时，需要收集该节点下所有的订阅者（可能还需要递归其子树，取决于语义）。

实现一个高效的通配符匹配器本身就是一个不小的挑战，需要考虑缓存、匹配性能等问题。对于初期，如果不需要，完全可以搁置。

4.3 性能考量：锁粒度、队列与线程模型

• 锁粒度：我们目前的实现用一把大锁（mutex_）保护了所有共享数据。在订阅/发布非常频繁的高并发场景下，这可能成为瓶颈。可以考虑进行锁拆分：
  • 用一把读写锁（std::shared_mutex）保护subscriptions_（读多写少）。
  • 用另一把互斥锁保护messageQueue_。 但这会显著增加复杂度，需要仔细处理跨锁的操作原子性。除非性能测试表明锁竞争是主要瓶颈，否则保持简单的一把锁是更稳妥的选择。
• 队列选择：我们使用了std::deque。std::queue（默认适配std::deque）也可以。在极端高性能场景，可以考虑无锁队列（如moodycamel::ConcurrentQueue），但这属于高级优化。
• 线程模型：我们使用了一个消费者线程。如果消息处理是计算密集型的，单个线程可能成为瓶颈。可以扩展为线程池模型：一个分发线程（或发布线程本身）将消息放入多个工作线程的队列，或者使用一个共享队列配合多个工作线程。这引入了消息顺序问题（不同消息可能被并行处理，顺序无法保证），需要根据业务需求权衡。

5. 实战示例与常见问题排查

让我们用一个简单的例子来演示如何使用这个PubSub类。

#include <iostream> #include <chrono> #include <thread> int main() { PubSub bus; // 订阅者1： 订阅 "news" auto handle1 = bus.subscribe("news", [](const std::string& topic, const std::any& msg) { try { auto& text = std::any_cast<const std::string&>(msg); std::cout << "[Subscriber1 on \"" << topic << "\"]: " << text << std::endl; } catch (const std::bad_any_cast&) { std::cout << "Wrong message type on topic: " << topic << std::endl; } }); // 订阅者2： 也订阅 "news" auto handle2 = bus.subscribe("news", [](const std::string& topic, const std::any& msg) { try { auto& text = std::any_cast<const std::string&>(msg); std::cout << "[Subscriber2 on \"" << topic << "\"]: " << text << std::endl; } catch (const std::bad_any_cast&) { // 处理类型错误 } }); // 订阅者3： 订阅 "weather" auto handle3 = bus.subscribe("weather", [](const std::string& topic, const std::any& msg) { try { auto& temp = std::any_cast<const double&>(msg); std::cout << "[Weather Report] Current temperature: " << temp << "°C" << std::endl; } catch (const std::bad_any_cast&) { // 处理类型错误 } }); // 发布消息 bus.publish("news", std::string("Breaking: C++20 is officially released!")); bus.publish("weather", 23.5); bus.publish("news", std::string("Update: Conference starts tomorrow.")); // 取消订阅者2 std::this_thread::sleep_for(std::chrono::milliseconds(100)); // 等待之前的消息处理完 bus.unsubscribe(handle2); std::cout << "\n--- Unsubscribed Subscriber2 ---\n" << std::endl; bus.publish("news", std::string("Last news: Workshop is full.")); // 给后台线程一点时间处理剩余消息 std::this_thread::sleep_for(std::chrono::milliseconds(200)); // PubSub bus 析构时会自动调用 stop() return 0; }

运行这个例子，你会看到Subscriber1和Subscriber2都收到了前两条新闻，但在取消Subscriber2后，只有Subscriber1收到了最后一条新闻。

常见问题与排查技巧：

1. 消息丢失：发布后订阅者没收到。
   • 检查点：订阅是否在发布之前完成？在异步模型中，如果订阅操作（修改subscriptions_）和发布操作（查询subscriptions_）之间没有正确的同步，可能会错过。我们的实现通过共用一把锁避免了这个问题。
   • 检查点：回调函数中是否有异常未被捕获？我们的workerThread已经做了捕获，但如果你的实现没有，一个异常会导致线程退出，后续消息全部丢失。
2. 内存泄漏：订阅者句柄未正确取消。
   • 最佳实践：使用RAII对象管理订阅生命周期。创建一个ScopedSubscription类，在构造函数中订阅，在析构函数中取消。

   class ScopedSubscription { PubSub& bus_; SubscriptionHandle handle_; public: ScopedSubscription(PubSub& bus, const std::string& topic, Callback cb) : bus_(bus), handle_(bus.subscribe(topic, std::move(cb))) {} ~ScopedSubscription() { bus_.unsubscribe(handle_); } // 禁止拷贝 ScopedSubscription(const ScopedSubscription&) = delete; ScopedSubscription& operator=(const ScopedSubscription&) = delete; // 允许移动 ScopedSubscription(ScopedSubscription&&) = default; ScopedSubscription& operator=(ScopedSubscription&&) = default; };

3. 程序卡死或崩溃：
   • 死锁：确保回调函数内部不会尝试去获取保护PubSub内部结构的同一个锁。我们通过复制订阅者列表避免了这一点。
   • 悬空回调：这是最常见的崩溃原因。确保在订阅者对象销毁前取消订阅。使用前面提到的weak_ptr或RAII Token方案来系统化解决。
   • 工作线程未正常退出：在析构函数中，必须确保running_标志被设置为false，并调用cv_.notify_all()来唤醒可能正在等待的工作线程，然后join()它。否则程序退出时，线程可能还在运行，访问已销毁的成员变量导致崩溃。
4. 性能瓶颈：
   • 锁竞争：使用性能分析工具（如perf,VTune）查看mutex_的争用情况。如果争用激烈，考虑拆分锁或使用无锁数据结构。
   • 队列积压：如果消息生产速度远大于消费速度，队列会无限增长。需要设计背压（Backpressure）策略，例如丢弃旧消息、阻塞发布者或提供队列满的通知。

实现一个健壮的、生产可用的C++ PubSub模型需要考虑诸多细节，但核心思想是清晰的：解耦、异步、安全。从这个小而美的核心开始，你可以根据项目的具体需求，逐步添加通配符、持久化、网络传输等功能，最终构建出属于你自己的强大消息通信基础设施。