【從零實現Json-Rpc框架】- 項目實現 - 服務端主題實現及整體封裝

📢博客主頁：https://blog.csdn.net/2301_779549673
📢博客倉庫：https://gitee.com/JohnKingW/linux_test/tree/master/lesson
📢歡迎點贊 👍 收藏 ?留言 📝 如有錯誤敬請指正！
📢本文由 JohnKi 原創，首發于 CSDN🙉
📢未來很長，值得我們全力奔赴更美好的生活?

在這里插入圖片描述

文章目錄

📢前言
🏳??🌈一、服務端主題管理模塊?
- 1.1 核心功能
- 1.2 核心設計思路
- 1.3 主題結構構造
- 1.4 訂閱者結構構造
- 1.5 主題和訂閱者統籌管理
- - (1) onTopicRequest
  - (2) onShutdown
  - (3) 細節實現方法
- 1.8 框架代碼
- 1.7 整體代碼
🏳??🌈二、服務端整體實現
- 2.1 邏輯框架
- 2.1.1 注冊中心（RegistryServer）?
- 2.1.2 RPC 服務核心（RpcServer）?
- - 2.1.3 主題服務（TopicServer）?
- 2.2 邏輯代碼
- 2.3 整體代碼
👥總結

📢前言

截至現在，在項目實現上，我們已經封裝好了 零碎接口，對 各種消息及常用結構體 進行了封裝

也實現了 dispatcher 路由轉發的功能。

在 服務端 方面，完成了

業務函數回調總結 - rpc_route.hpp
服務的提供、發現、注冊 - rpc_registry.hpp

在 客戶端 方面，完成了

消息請求及其回調 - requestor.hpp
消息請求發送 - rpc_caller.hpp

服務端 未完成部分

服務端主題管理模塊 - rpc_topic.hpp
服務端功能整合 - rpc_server.hpp

客戶端 未完成部分

客戶的提供、發現、注冊 - rpc_registry.hpp
客戶端主題管理模塊 - rpc_topic.hpp
客戶端功能整合 - rpc_client.hpp

這一篇文章，筆者就為服務端的 主題實現 和 整體封裝 畫上句號

🏳??🌈一、服務端主題管理模塊?

1.1 核心功能

這段代碼實現了一個 ?服務端主題管理模塊?（TopicManager），支持 ?發布-訂閱模式 的核心功能，包括：

?主題的創建與刪除
客戶端的訂閱與取消訂閱
消息的發布與推送
連接斷開時的自動清理

1.2 核心設計思路

?線程安全：通過 std::mutex 保護共享數據（_topics 和 _subscribers）。
?數據映射：

_topics：維護主題名稱到 Topic 對象的映射。
_subscribers：維護客戶端連接到 Subscriber 對象的映射。

?操作統一入口：通過 onTopicRequest 分發不同類型的主題操作請求。

1.3 主題結構構造

作用：表示一個主題，管理其所有訂閱者。
?成員：

_subscribers：存儲所有訂閱者的集合（Subscriber::ptr）。

?方法：

appendSubscriber / removeSubscriber：增刪訂閱者。
pushMessage：向所有訂閱者發送消息。

// 主題名稱 和 其訂閱者連接 的映射
struct Topic {using ptr = std::shared_ptr<Topic>;std::mutex _mutex;std::string _topic_name;std::unordered_set<Subscriber::ptr> _subscribers; // 當前主題的訂閱者的連接Topic(const std::string& topic_name) : _topic_name(topic_name) {}// 增加訂閱者void appendSubscriber(const Subscriber::ptr& subscriber);// 刪除訂閱者void removeSubscriber(const Subscriber::ptr& subscriber);// 推送消息void pushMessage(const BaseMessage::ptr& msg);
};

1.4 訂閱者結構構造

作用：表示一個訂閱者，記錄其訂閱的主題。
?成員：

_conn：訂閱者的網絡連接對象。
_topics：訂閱者當前訂閱的所有主題名稱集合。

?方法：

appendTopic / removeTopic：增刪訂閱的主題。

// 定義一個訂閱者對象，記錄其訂閱的 所有主題名稱
struct Subscriber {using ptr = std::shared_ptr<Subscriber>;std::mutex _mutex;BaseConnection::ptr _conn;std::unordered_set<std::string> _topics; // 該訂閱者所訂閱的主題名稱Subscriber(const BaseConnection::ptr& conn) : _conn(conn) {};// 增加訂閱的主題void appendTopic(const std::string& topic);// 刪除訂閱的主題void removeTopic(const std::string& topic);
};

1.5 主題和訂閱者統籌管理

我們需要建立兩個映射關系

主題名 -> 主題結構
訂閱者連接 -> 訂閱者結構

std::mutex _mutex;
std::unordered_map<std::string, Topic::ptr> _topics;
std::unordered_map<BaseConnection::ptr, Subscriber::ptr> _subscribers;

核心接口解析

(1) onTopicRequest

?功能：處理客戶端發送的主題操作請求（總入口）。
?操作類型：

TOPIC_CREATE：調用 topicCreate 創建主題。
TOPIC_REMOVE：調用 topicRemove 刪除主題。
TOPIC_SUBSCRIBE：調用 topicSubscribe 訂閱主題。
TOPIC_CANCEL：調用 topicCancel 取消訂閱。
TOPIC_PUBLISH：調用 topicPublish 發布消息。
錯誤處理：若操作失敗，返回錯誤響應（errorResponse）。

(2) onShutdown

?功能：處理客戶端連接斷開時的清理邏輯。
?步驟：

從 _subscribers 中移除訂閱者。
遍歷訂閱者的所有主題，從主題的訂閱列表中移除該訂閱者。

(3) 細節實現方法

變更類

// 構造一個主題對象，添加映射關系的管理
void topicCreate(const BaseConnection::ptr& conn, const TopicRequest::ptr& msg);// 刪除一個主題對象，刪除映射關系的管理
void topicRemove(const BaseConnection::ptr& conn, const TopicRequest::ptr& msg);// 將訂閱者訂閱到指定主題。
bool topicSubscribe(const BaseConnection::ptr& conn, const TopicRequest::ptr& msg);// 取消該主題的訂閱者
void topicCancel(const BaseConnection::ptr& conn, const TopicRequest::ptr& msg);// 向各個該主題的 訂閱者 發布主題消息
bool topicPublish(const BaseConnection::ptr& conn, const TopicRequest::ptr& msg);

返回類

//  返回一個錯誤響應
void errorResponse(const BaseConnection::ptr& conn, const TopicRequest::ptr& msg, RCode rcode);//  返回一個主題響應
void topicResponse(const BaseConnection::ptr& conn, const TopicRequest::ptr& msg);

1.8 框架代碼

#pragma once
#include "../common/net.hpp"
#include "../common/message.hpp"
#include <unordered_set>namespace rpc
{namespace server{class TopicManager{public:using ptr = std::shared_ptr<TopicManager>;// 統一處理有關主題的請求void onTopicRequest(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg);// 關閉連接void onShutdown(const BaseConnection::ptr &conn);private://  返回一個錯誤響應void errorResponse(const BaseConnection::ptr& conn, const TopicRequest::ptr& msg, RCode rcode);//  返回一個主題響應void topicResponse(const BaseConnection::ptr& conn, const TopicRequest::ptr& msg);private:// 構造一個主題對象，添加映射關系的管理void topicCreate(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg);// 刪除一個主題對象，刪除映射關系的管理void topicRemove(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg);// 將訂閱者訂閱到指定主題。bool topicSubscribe(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg);// 取消該主題的訂閱者 void topicCancel(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg);// 向各個該主題的 訂閱者 發布主題消息bool topicPublish(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg);private:// 定義一個訂閱者對象，記錄其訂閱的 所有主題名稱struct Subscriber{using ptr = std::shared_ptr<Subscriber>;std::mutex _mutex;BaseConnection::ptr _conn;std::unordered_set<std::string> _topics; // 該訂閱者所訂閱的主題名稱Subscriber(const BaseConnection::ptr &conn) : _conn(conn) {};// 增加訂閱的主題void appendTopic(const std::string &topic);// 刪除訂閱的主題void removeTopic(const std::string &topic);};// 主題名稱 和 其訂閱者連接 的映射struct Topic{using ptr = std::shared_ptr<Topic>;std::mutex _mutex;std::string _topic_name;std::unordered_set<Subscriber::ptr> _subscribers; // 當前主題的訂閱者的連接Topic(const std::string &topic_name) : _topic_name(topic_name) {}// 增加訂閱者void appendSubscriber(const Subscriber::ptr &subscriber);// 刪除訂閱者void removeSubscriber(const Subscriber::ptr &subscriber);// 推送消息void pushMessage(const BaseMessage::ptr &msg);};private:std::mutex _mutex;std::unordered_map<std::string, Topic::ptr> _topics;std::unordered_map<BaseConnection::ptr, Subscriber::ptr> _subscribers;};}
}

1.7 整體代碼

#pragma once
#include "../common/net.hpp"
#include "../common/message.hpp"
#include <unordered_set>namespace rpc
{namespace server{class TopicManager{public:using ptr = std::shared_ptr<TopicManager>;// 統一處理有關主題的請求void onTopicRequest(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg){TopicOptype topic_optype = msg->optype();bool ret = true;switch(topic_optype){case TopicOptype::TOPIC_CREATE: topicCreate(conn, msg); break;              // 主題創建case TopicOptype::TOPIC_REMOVE: topicRemove(conn, msg); break;              // 主題刪除case TopicOptype::TOPIC_SUBSCRIBE: ret = topicSubscribe(conn, msg); break;  // 主題訂閱case TopicOptype::TOPIC_CANCEL: topicCancel(conn, msg); break;              // 主題取消訂閱case TopicOptype::TOPIC_PUBLISH: ret = topicPublish(conn, msg); break;      // 主題發布default: return errorResponse(conn, msg, RCode::RCODE_INVALID_OPTYPE); break;}if(!ret) return errorResponse(conn, msg, RCode::RCODE_NOT_FOUND_TOPIC);return topicResponse(conn, msg);}// 關閉連接void onShutdown(const BaseConnection::ptr &conn){// 消息發布者斷開連接，不需要任何操作// 1. 判斷斷開連接的是否為訂閱者，不是的話直接返回// 2. 獲取到訂閱者退出，受影響的主題對象// 3. 從主題對象中，移除訂閱者// 4. 從訂閱者映射信息中，刪除訂閱者std::vector<Topic::ptr> topics;Subscriber::ptr subscriber;{std::unique_lock<std::mutex> lock(_mutex);auto sub_it = _subscribers.find(conn);if(sub_it == _subscribers.end()){ELOG("該訂閱者連接不存在");return;}// 獲取該訂閱者鎖定月的主題subscriber = sub_it->second;// 2. 獲取到訂閱者退出，受影響的主題對象for(auto& topic_name : subscriber->_topics){auto topic_it = _topics.find(topic_name);if(topic_it == _topics.end())continue;topics.push_back(topic_it->second);}_subscribers.erase(sub_it);}}private:void errorResponse(const BaseConnection::ptr& conn, const TopicRequest::ptr& msg, RCode rcode){auto msg_rsp = MessageFactory::create<TopicResponse>();msg_rsp->setId(msg->rid());msg_rsp->setMType(MType::RSP_TOPIC);msg_rsp->setRcode(rcode);return conn->send(msg_rsp);}void topicResponse(const BaseConnection::ptr& conn, const TopicRequest::ptr& msg){auto msg_rsp = MessageFactory::create<TopicResponse>();msg_rsp->setId(msg->rid());msg_rsp->setMType(MType::RSP_TOPIC);msg_rsp->setRcode(RCode::RCODE_OK);return conn->send(msg_rsp);}private:// 構造一個主題對象，添加映射關系的管理void topicCreate(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg){std::unique_lock<std::mutex> lock(_mutex);// 獲取主題名字std::string topic_name = msg->topicKey();// 構造一個主題對象auto topic = std::make_shared<Topic>(topic_name);// 增加訂閱者_topics.insert(std::make_pair(topic_name, topic));std::cout << "創建主題" <<  topic_name << std::endl;}// 刪除一個主題對象，刪除映射關系的管理void topicRemove(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg){// 1. 查看當前主題，有哪些訂閱者，然后從訂閱者中將主題信息刪掉// 2. 刪除主題的數據 -- 主題名稱余出題對象的映射std::string topic_name = msg->topicKey();std::unordered_set<Subscriber::ptr> subscribers; // 記錄 當前主題 的 所有訂閱者連接{std::unique_lock<std::mutex> lock(_mutex);auto it = _topics.find(msg->topicKey());if (it == _topics.end()){ELOG("沒有找到 %s 主題的訂閱者", msg->topicKey().c_str());return;}subscribers = it->second->_subscribers;_topics.erase(it); // 刪除主題對象}for (auto &subscriber : subscribers){subscriber->removeTopic(topic_name);}}// 將訂閱者訂閱到指定主題。bool topicSubscribe(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg){Topic::ptr topic;Subscriber::ptr subscriber;{std::unique_lock<std::mutex> lock(_mutex);// 1. 查找或創建訂閱者對象（Subscriber）auto topic_it = _topics.find(msg->topicKey());if (topic_it == _topics.end()){return false;}topic = topic_it->second;auto sub_it = _subscribers.find(conn);if (sub_it != _subscribers.end()){subscriber = sub_it->second;}else{subscriber = std::make_shared<Subscriber>(conn);_subscribers.insert(std::make_pair(conn, subscriber));}}// 2. 在主題對象中，新增一個訂閱者對象關聯的連接；  在訂閱者對象中新增一個訂閱的主題topic->appendSubscriber(subscriber);subscriber->appendTopic(msg->topicKey());return true;}// 取消該主題的訂閱者 void topicCancel(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg){// 1. 先找出主題對象，和訂閱者對象Topic::ptr topic;Subscriber::ptr subscriber;{std::unique_lock<std::mutex> lock(_mutex);auto topic_it = _topics.find(msg->topicKey());if (topic_it != _topics.end()){topic = topic_it->second;}auto sub_it = _subscribers.find(conn);if (sub_it != _subscribers.end()){subscriber = sub_it->second;}}// 2. 從主對象中刪除當前的訂閱者連接if(subscriber)subscriber->removeTopic(msg->topicKey());if(topic && subscriber)topic->removeSubscriber(subscriber);}// 向各個該主題的 訂閱者 發布主題消息bool topicPublish(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg){Topic::ptr topic;{std::unique_lock<std::mutex> lock(_mutex);auto topic_it = _topics.find(msg->topicKey());if (topic_it == _topics.end()){ELOG("沒有找到 %s 主題的訂閱者", msg->topicKey().c_str());return false;}topic = topic_it->second;}topic->pushMessage(msg);return true;}private:// 定義一個訂閱者對象，記錄其訂閱的 所有主題名稱struct Subscriber{using ptr = std::shared_ptr<Subscriber>;std::mutex _mutex;BaseConnection::ptr _conn;std::unordered_set<std::string> _topics; // 該訂閱者所訂閱的主題名稱Subscriber(const BaseConnection::ptr &conn) : _conn(conn) {};// 增加訂閱的主題void appendTopic(const std::string &topic){std::unique_lock<std::mutex> lock(_mutex);_topics.insert(topic);}void removeTopic(const std::string &topic){std::unique_lock<std::mutex> lock(_mutex);_topics.erase(topic);}};// 主題名稱 和 其訂閱者連接 的映射struct Topic{using ptr = std::shared_ptr<Topic>;std::mutex _mutex;std::string _topic_name;std::unordered_set<Subscriber::ptr> _subscribers; // 當前主題的訂閱者的連接Topic(const std::string &topic_name) : _topic_name(topic_name) {}// 增加訂閱者void appendSubscriber(const Subscriber::ptr &subscriber){std::unique_lock<std::mutex> lock(_mutex);_subscribers.insert(subscriber);}// 刪除訂閱者void removeSubscriber(const Subscriber::ptr &subscriber){std::unique_lock<std::mutex> lock(_mutex);_subscribers.erase(subscriber);}// 推送消息void pushMessage(const BaseMessage::ptr &msg){std::unique_lock<std::mutex> lock(_mutex);for (auto &subscriber : _subscribers){subscriber->_conn->send(msg);}}};private:std::mutex _mutex;std::unordered_map<std::string, Topic::ptr> _topics;std::unordered_map<BaseConnection::ptr, Subscriber::ptr> _subscribers;};}
}

🏳??🌈二、服務端整體實現

2.1 邏輯框架

一個 ?分布式 RPC 服務端系統

包含三個核心模塊：?服務注冊中心、RPC 服務核心 和 ?主題服務。

以下是各模塊的作用及整體架構：

2.1.1 注冊中心（RegistryServer）?

?功能：

?服務注冊與發現：接收服務提供者（Provider）注冊的服務信息，供消費者（Consumer）查詢可用服務。
連接管理：在客戶端斷開時清理相關資源（如服務下線通知）。

?關鍵成員：

_pd_manager：管理服務注冊與發現的業務邏輯（如維護服務列表）。
_dispatcher：分發客戶端請求到對應的處理邏輯。
_server：底層網絡服務器，監聽端口并處理連接。

?使用場景：

服務提供者啟動時向注冊中心注冊自身服務。
消費者通過注冊中心查詢可用的服務地址和方法。

2.1.2 RPC 服務核心（RpcServer）?

?功能：

?RPC 服務管理：啟動 RPC 服務，處理客戶端調用請求。
服務注冊（可選）?：將服務方法注冊到注冊中心（若啟用）。
?請求路由：將 RPC 請求路由到對應的業務處理邏輯。

?關鍵成員：

_router：路由請求到具體的服務方法（如根據方法名匹配處理函數）。
_req_client：與注冊中心通信的客戶端（用于服務注冊或發現）。
_dispatcher：分發網絡消息到業務邏輯。
?
配置選項：
_enableRegistry：是否啟用注冊中心（決定是否自動注冊服務）。

2.1.3 主題服務（TopicServer）?

?功能：

?發布-訂閱模式：管理主題的創建、訂閱和消息推送。
?消息廣播：向訂閱特定主題的客戶端推送消息。

?關鍵成員：

_topic_manager：管理主題和訂閱者（如 TopicManager 類）。
_dispatcher：處理客戶端的訂閱/發布請求。

2.2 邏輯代碼

#pragma once#include "../common/dispatcher.hpp"
#include "../client/rpc_client.hpp"#include "rpc_route.hpp"
#include "rpc_registry.hpp"
#include "rpc_topic.hpp"namespace rpc
{namespace server{// 注冊中心的服務端實現// 啟動服務：監聽指定端口，接收客戶端（服務提供者/發現者）連接。// ?消息路由：將服務注冊/發現請求分發給業務處理器（PDManager）。// ?連接管理：在客戶端斷開時清理相關資源（如服務下線通知）。class RegistryServer{// 注冊中心服務端，只需要針對服務注冊與發現請求進行處理即可public:using ptr = std::shared_ptr<RegistryServer>;RegistryServer(int port);void start();private:void onConnShutdown(const BaseConnection::ptr &conn);private:ProviderDiscovererManager::ptr _pd_manager;Dispatcher::ptr _dispatcher;BaseServer::ptr _server;};// ?RPC 服務端核心類，負責管理 RPC 服務的生命周期// 啟動 RPC 服務：監聽指定端口，處理客戶端 RPC 請求。// ?服務注冊?（可選）：將服務方法注冊到注冊中心，供客戶端發現。// ?請求路由：將接收到的 RPC 請求分發給對應的業務處理邏輯。class RpcServer{public:using ptr = std::shared_ptr<RpcServer>;RpcServer(const Address &access_addr, bool enableRegistry = false, const Address &registry_server_addr = Address());// 注冊服務到注冊中心void registerMethod(const ServiceDescribe::ptr &service);void start();private:bool _enableRegistry;Address _access_addr;rpc::client::RegistryClient::ptr _req_client;RpcRouter::ptr _router;Dispatcher::ptr _dispatcher;BaseServer::ptr _server;};class TopicServer{public:using ptr = std::shared_ptr<TopicServer>;TopicServer(int port);void start();private:void onConnShutdown(const BaseConnection::ptr &conn);private:TopicManager::ptr _topic_manager;Dispatcher::ptr _dispatcher;BaseServer::ptr _server;};}
}

2.3 整體代碼

#pragma once#include "../common/dispatcher.hpp"
#include "../client/rpc_client.hpp" #include "rpc_route.hpp"
#include "rpc_registry.hpp"
#include "rpc_topic.hpp"namespace rpc{namespace server{// 注冊中心的服務端實現// 啟動服務：監聽指定端口，接收客戶端（服務提供者/發現者）連接。// ?消息路由：將服務注冊/發現請求分發給業務處理器（PDManager）。// ?連接管理：在客戶端斷開時清理相關資源（如服務下線通知）。class RegistryServer{// 注冊中心服務端，只需要針對服務注冊與發現請求進行處理即可public:using ptr = std::shared_ptr<RegistryServer>;RegistryServer(int port): _pd_manager(std::make_shared<ProviderDiscovererManager>()),_dispatcher(std::make_shared<Dispatcher>()){ // 1. 注冊服務請求處理器// 將 PDManager::onServiceRequest 綁定到 MType::REQ_SERVICE 消息類型。當收到服務注冊或發現請求時，調用此方法處理auto service_cb = std::bind(&ProviderDiscovererManager::onServiceRequest, _pd_manager.get(), std::placeholders::_1, std::placeholders::_2);_dispatcher->registerHandler<ServiceRequest>(MType::REQ_SERVICE, service_cb);// 2. 創建底層服務器并設置回調// 通過 ServerFactory 創建底層服務器，設置消息總入口為 Dispatcher::onMessage_server = rpc::ServerFactory::create(port);auto message_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(), std::placeholders::_1, std::placeholders::_2);_server->setMessageCallback(message_cb);// 3. 設置連接關閉回調// 當客戶端斷開連接時，調用 onConnShutdown 清理相關資源auto close_cb = std::bind(&RegistryServer::onConnShutdown, this, std::placeholders::_1);_server->setCloseCallback(close_cb);}void start(){_server->start();   // 啟動服務器，開始監聽端口}private:void onConnShutdown(const BaseConnection::ptr& conn){_pd_manager->onConnShutdown(conn);  // 通知 PDManager 處理連接斷開}private:ProviderDiscovererManager::ptr _pd_manager;Dispatcher::ptr _dispatcher;BaseServer::ptr _server;};// ?RPC 服務端核心類，負責管理 RPC 服務的生命周期// 啟動 RPC 服務：監聽指定端口，處理客戶端 RPC 請求。// ?服務注冊?（可選）：將服務方法注冊到注冊中心，供客戶端發現。// ?請求路由：將接收到的 RPC 請求分發給對應的業務處理邏輯。class RpcServer{public:using ptr = std::shared_ptr<RpcServer>;RpcServer(const Address& access_addr, bool enableRegistry = false, const Address& registry_server_addr = Address()): _enableRegistry(enableRegistry),_access_addr(access_addr),_router(std::make_shared<rpc::server::RpcRouter>()),_dispatcher(std::make_shared<Dispatcher>()){// 1. ?創建注冊客戶端?（若啟用注冊）if(enableRegistry){_req_client = std::make_shared<client::RegistryClient>(registry_server_addr.first, registry_server_addr.second);}// 2. 注冊 RPC 請求處理回調auto rpc_cb = std::bind(&RpcRouter::onRpcRequest, _router.get(), std::placeholders::_1, std::placeholders::_2);_dispatcher->registerHandler<RpcRequest>(MType::REQ_RPC, rpc_cb);// 3. 創建底層服務器并設置回調_server = rpc::ServerFactory::create(access_addr.second);auto message_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(), std::placeholders::_1, std::placeholders::_2);_server->setMessageCallback(message_cb);}// 注冊服務到注冊中心void registerMethod(const ServiceDescribe::ptr& service){if(_enableRegistry)_req_client->registryMethod(service->method(), _access_addr);_router->registerMethod(service);}void start(){_server->start();}private:bool _enableRegistry;Address _access_addr;rpc::client::RegistryClient::ptr _req_client;RpcRouter::ptr _router;Dispatcher::ptr _dispatcher;BaseServer::ptr _server;};class TopicServer{public:using ptr = std::shared_ptr<TopicServer>;TopicServer(int port): _topic_manager(std::make_shared<TopicManager>()),_dispatcher(std::make_shared<Dispatcher>()){// 1. 注冊主題請求處理器auto topic_cb = std::bind(&TopicManager::onTopicRequest, _topic_manager.get(), std::placeholders::_1, std::placeholders::_2);_dispatcher->registerHandler<TopicRequest>(MType::REQ_TOPIC, topic_cb);// 2. 創建底層服務器并設置回調_server = ServerFactory::create(port);auto message_cb = std::bind(&Dispatcher::onMessage, _dispatcher.get(), std::placeholders::_1, std::placeholders::_2);_server->setMessageCallback(message_cb);// 3. 設置連接關閉回調auto close_cb = std::bind(&TopicServer::onConnShutdown, this, std::placeholders::_1);_server->setCloseCallback(close_cb);}void start(){_server->start();}private:void onConnShutdown(const BaseConnection::ptr& conn){_topic_manager->onShutdown(conn);}private:TopicManager::ptr _topic_manager;Dispatcher::ptr _dispatcher;BaseServer::ptr _server;};}
}