如何基于STL实现更通用的C++发布-订阅模式？

核心思路 #

利用std::function和lambda捕获智能指针替代传统接口类继承，实现灵活的回调绑定；通过模板支持任意参数数量与类型；用std::async实现异步转发；单独封装查询类处理一对一请求-响应场景。

1. 存储订阅者与成员函数的组合并完成注册 #

发布者通过模板成员函数接收订阅者智能指针和成员函数指针，将二者绑定为std::function对象存储。lambda捕获智能指针可保证订阅者生命周期与回调绑定周期一致，避免悬空指针问题。

关键实现代码：

template<typename... Args>
class Publisher {
private:
    using Callback = std::function<void(Args...)>;
    std::unordered_map<std::string, std::vector<Callback>> channels_;

public:
    // 注册订阅者：智能指针 + 成员函数指针
    template<typename SubscriberType>
    void subscribe(const std::string& channel, 
                   std::shared_ptr<SubscriberType> subscriber, 
                   void(SubscriberType::*callback)(Args...)) {
        channels_[channel].emplace_back(
            [subscriber, callback](Args... args) {
                // 完美转发参数调用成员函数
                (subscriber.get()->*callback)(std::forward<Args>(args)...);
            }
        );
    }
};

2. 发布时绑定具体参数 #

发布时遍历对应通道的回调列表，通过完美转发将参数传递给每个回调，保证参数的原始值类别（左值/右值），避免不必要的拷贝。

添加发布方法到Publisher类：

void publish(const std::string& channel, Args... args) {
    auto channel_it = channels_.find(channel);
    if (channel_it == channels_.end()) return;

    for (auto& callback : channel_it->second) {
        callback(std::forward<Args>(args)...);
    }
}

3. 扩展为异步版本 #

利用std::async以异步方式启动每个回调任务，无需手动管理线程。若需等待异步任务完成，可返回std::future集合；若只需后台执行，直接启动异步任务即可。

添加异步发布方法到Publisher类：

void publish_async(const std::string& channel, Args... args) {
    auto channel_it = channels_.find(channel);
    if (channel_it == channels_.end()) return;

    for (auto& callback : channel_it->second) {
        // 异步启动回调，无需等待结果
        std::async(std::launch::async, callback, std::forward<Args>(args)...);
    }
}

4. 实现查询版本（请求-响应模式） #

查询模式为一对一的请求-响应场景，需存储单一提供者的回调并返回结果。同样用std::function绑定智能指针与成员函数，支持同步/异步查询。

实现查询提供者类：

template<typename ReturnType, typename... Args>
class QueryProvider {
private:
    using ProviderCallback = std::function<ReturnType(Args...)>;
    ProviderCallback provider_;

public:
    // 注册提供者：智能指针 + 返回值型成员函数指针
    template<typename ProviderType>
    void register_provider(std::shared_ptr<ProviderType> provider, 
                           ReturnType(ProviderType::*callback)(Args...)) {
        provider_ = [provider, callback](Args... args) -> ReturnType {
            return (provider.get()->*callback)(std::forward<Args>(args)...);
        };
    }

    // 同步查询
    ReturnType query(Args... args) {
        if (!provider_) {
            throw std::runtime_error("No provider registered for query");
        }
        return provider_(std::forward<Args>(args)...);
    }

    // 异步查询，返回future获取结果
    std::future<ReturnType> query_async(Args... args) {
        if (!provider_) {
            throw std::runtime_error("No provider registered for query");
        }
        return std::async(std::launch::async, provider_, std::forward<Args>(args)...);
    }
};

完整程序实现与测试 #

#include <iostream>
#include <memory>
#include <functional>
#include <unordered_map>
#include <vector>
#include <future>
#include <stdexcept>
#include <chrono>
#include <thread>

// 通用发布者类
template<typename... Args>
class Publisher {
private:
    using Callback = std::function<void(Args...)>;
    std::unordered_map<std::string, std::vector<Callback>> channels_;

public:
    template<typename SubscriberType>
    void subscribe(const std::string& channel, 
                   std::shared_ptr<SubscriberType> subscriber, 
                   void(SubscriberType::*callback)(Args...)) {
        channels_[channel].emplace_back(
            [subscriber, callback](Args... args) {
                (subscriber.get()->*callback)(std::forward<Args>(args)...);
            }
        );
    }

    void publish(const std::string& channel, Args... args) {
        auto channel_it = channels_.find(channel);
        if (channel_it == channels_.end()) return;

        for (auto& callback : channel_it->second) {
            callback(std::forward<Args>(args)...);
        }
    }

    void publish_async(const std::string& channel, Args... args) {
        auto channel_it = channels_.find(channel);
        if (channel_it == channels_.end()) return;

        for (auto& callback : channel_it->second) {
            std::async(std::launch::async, callback, std::forward<Args>(args)...);
        }
    }
};

// 通用查询提供者类
template<typename ReturnType, typename... Args>
class QueryProvider {
private:
    using ProviderCallback = std::function<ReturnType(Args...)>;
    ProviderCallback provider_;

public:
    template<typename ProviderType>
    void register_provider(std::shared_ptr<ProviderType> provider, 
                           ReturnType(ProviderType::*callback)(Args...)) {
        provider_ = [provider, callback](Args... args) -> ReturnType {
            return (provider.get()->*callback)(std::forward<Args>(args)...);
        };
    }

    ReturnType query(Args... args) {
        if (!provider_) {
            throw std::runtime_error("No provider registered for query");
        }
        return provider_(std::forward<Args>(args)...);
    }

    std::future<ReturnType> query_async(Args... args) {
        if (!provider_) {
            throw std::runtime_error("No provider registered for query");
        }
        return std::async(std::launch::async, provider_, std::forward<Args>(args)...);
    }
};

// 测试用订阅者类
class MessageSubscriber {
public:
    void receiveInChannel1(const std::string& msg, int priority) {
        std::cout << "Subscriber 1 received: [" << priority << "] " << msg << std::endl;
    }
};

// 测试用查询提供者类
class ConfigProvider {
public:
    std::string getConfigValue(const std::string& key) {
        if (key == "log_level") return "INFO";
        if (key == "max_connection") return "100";
        return "unknown";
    }
};

int main() {
    // 测试发布-订阅
    Publisher<const std::string&, int> msg_publisher;
    auto subscriber = std::make_shared<MessageSubscriber>();
    msg_publisher.subscribe("channel1", subscriber, &MessageSubscriber::receiveInChannel1);

    std::cout << "=== 同步发布 ===" << std::endl;
    msg_publisher.publish("channel1", "Hello Sync", 1);

    std::cout << "=== 异步发布 ===" << std::endl;
    msg_publisher.publish_async("channel1", "Hello Async", 2);
    // 等待异步任务完成（仅用于测试，实际业务无需）
    std::this_thread::sleep_for(std::chrono::seconds(1));

    // 测试查询
    QueryProvider<std::string, const std::string&> config_query;
    auto config_provider = std::make_shared<ConfigProvider>();
    config_query.register_provider(config_provider, &ConfigProvider::getConfigValue);

    std::cout << "=== 同步查询 ===" << std::endl;
    try {
        std::cout << "log_level: " << config_query.query("log_level") << std::endl;
    } catch (const std::exception& e) {
        std::cerr << e.what() << std::endl;
    }

    std::cout << "=== 异步查询 ===" << std::endl;
    try {
        auto future_val = config_query.query_async("max_connection");
        std::cout << "max_connection: " << future_val.get() << std::endl;
    } catch (const std::exception& e) {
        std::cerr << e.what() << std::endl;
    }

    return 0;
}

内容的提问来源于stack exchange，提问作者Jörg Brüggmann