C++14/17发布订阅应用：非模板Broker存储多类型模板Queue的设计求解

我正在用C14/17编写一个小型发布订阅（Pub/Sub）应用用于练习和自学。标注14/17是因为目前仅使用了C17的std::scoped_lock特性。

在这个示例应用中，发布者和订阅者处于同一操作系统进程（同一编译二进制文件）中。这种情况下，合理的设计是用一个单独的类存储消息，比如使用std::unordered_map<std::string, std::deque>。我计划在main函数中实例化该类，并将其传递给发布者和订阅者的构造函数。

但当我尝试将消息存入自定义的模板队列类（支持不同类型消息，例如protobuf）时，问题出现了。参考以下代码：

// T 代表不同的protobuf类
template <class T>
class Queue {
public:
    std::mutex mutex_;

    void enqueueMessage(const T* message);
    const T* dequeueMessage();
    const int length() { return messages_.size();};
private:
    std::string id_;
    std::deque<const T*> messages_;
};

class Node
{
public:
    Node();
    template <class T>
    Publisher<T>* createPublisher(std::string const topic_name, Broker broker);
};

class Broker {
    public:
        template <class T>
        Publisher<T>* createPublisher(std::string const topic_name);
    private:
        /** 问题出在这里 **/
        std::unordered_map<std::string, Queue<T*>*> queues_;
};

int main(int argc, char** argv)
{
    // 该对象持有全局状态，将在需要时传递
    auto broker = std::make_shared<Broker>();

    EmployeeMessage empMessage = EmployeeMessage(/* 参数 */);
    WeatherMessage  weatherMessage = WeatherMessage(/* 参数 */);
    
    auto nodeEmp = std::make_shared<Node>();
    auto nodeWeather = std::make_shared<Node>();

    nodeEmp.createPublisher<EmployeeMessage>("name1", broker);
    nodeWeather.createPublisher<EmployeeMessage>("name2", broker);
}

Broker类的queues_成员无法使用类型T，因为Broker并非模板类。我不能将Broker改为模板类，否则每种消息类型都会对应一个Broker实例。该如何修正这个设计？

核心思路是类型擦除，让Broker能存储不同类型的Queue实例，同时保证类型安全。以下是两种可行的实现方式：

方式一：抽象基类+模板派生类 #

1. 定义非模板的抽象基类BaseQueue，提供虚析构函数（确保派生类对象能正确销毁）。
2. 让模板队列类Queue<T>继承自BaseQueue，实现具体的入队、出队逻辑。
3. Broker中用std::unordered_map<std::string, std::unique_ptr<BaseQueue>>存储队列，以此容纳不同类型的Queue实例。
4. 在Broker的模板方法createPublisher<T>中，检查topic对应的队列是否存在，不存在则创建；存在则通过dynamic_cast验证类型匹配，再返回对应类型的队列指针给Publisher。

修改后的关键代码示例：

// 抽象基类
class BaseQueue {
public:
    virtual ~BaseQueue() = default; // 必须有虚析构函数
};

// 模板队列类继承基类
template <class T>
class Queue : public BaseQueue {
public:
    std::mutex mutex_;

    void enqueueMessage(const T* message) {
        std::scoped_lock lock(mutex_);
        messages_.push_back(message);
    }

    const T* dequeueMessage() {
        std::scoped_lock lock(mutex_);
        if (messages_.empty()) return nullptr;
        auto msg = messages_.front();
        messages_.pop_front();
        return msg;
    }

    const int length() { 
        std::scoped_lock lock(mutex_);
        return messages_.size();
    };
private:
    std::string id_;
    std::deque<const T*> messages_;
};

class Broker {
public:
    template <class T>
    Publisher<T>* createPublisher(std::string const topic_name) {
        auto it = queues_.find(topic_name);
        if (it == queues_.end()) {
            // 创建对应类型的队列并插入map
            queues_[topic_name] = std::make_unique<Queue<T>>();
        } else {
            // 验证已有队列的类型匹配
            auto queue = dynamic_cast<Queue<T>*>(it->second.get());
            if (!queue) {
                throw std::runtime_error("该topic已存在，且消息类型不匹配");
            }
        }
        // 获取队列指针并创建Publisher
        auto queue_ptr = dynamic_cast<Queue<T>*>(queues_[topic_name].get());
        return new Publisher<T>(topic_name, queue_ptr);
    }
private:
    std::unordered_map<std::string, std::unique_ptr<BaseQueue>> queues_;
};

方式二：使用C++17的std::any存储队列 #

利用C++17引入的std::any直接存储任意类型的队列实例，Broker的map定义为std::unordered_map<std::string, std::any>，使用时通过std::any_cast取出对应类型的队列。

关键代码示例：

#include <any> // 需包含头文件

class Broker {
public:
    template <class T>
    Publisher<T>* createPublisher(std::string const topic_name) {
        auto it = queues_.find(topic_name);
        if (it == queues_.end()) {
            queues_[topic_name] = std::make_unique<Queue<T>>();
        } else {
            // 尝试取出对应类型的队列，验证类型匹配
            auto queue = std::any_cast<std::unique_ptr<Queue<T>>>(&it->second);
            if (!queue) {
                throw std::runtime_error("该topic已存在，且消息类型不匹配");
            }
        }
        // 获取队列指针并创建Publisher
        auto& queue_any = queues_[topic_name];
        auto queue_ptr = std::any_cast<std::unique_ptr<Queue<T>>>(&queue_any)->get();
        return new Publisher<T>(topic_name, queue_ptr);
    }
private:
    std::unordered_map<std::string, std::any> queues_;
};

额外注意事项 #

• 消息生命周期管理：当前代码存储原始指针，需确保消息对象的生命周期长于队列，或改用std::shared_ptr<T>管理消息，避免悬空指针。
• 线程安全：所有队列操作（包括length()）都要用std::scoped_lock包裹，避免数据竞争。
• 错误处理：类型不匹配时要抛出异常或返回错误，避免未定义行为。

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