当前位置： 首页 > news >正文

【从零实现Json-Rpc框架】- 项目实现 - 服务端主题实现及整体封装

news 2025/4/5 19:56:04

📢博客主页：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;};}
}

👥总结

本篇博文对 从零实现Json-Rpc框架】- 项目实现 - 服务端主题实现及整体封装 做了一个较为详细的介绍，不知道对你有没有帮助呢

觉得博主写得还不错的三连支持下吧！会继续努力的~

请添加图片描述


http://www.mrgr.cn/news/97013.html

相关文章：

  • 前端Uniapp接入UviewPlus详细教程！！！
  • [Linux]从零开始的vs code交叉调试arm Linux程序教程
  • 容器的CPU
  • CAD插入属性块 弹窗提示输入属性值——CAD知识讲堂
  • GenerationMixin：_sample方法(GenerationMode.SAMPLE, GenerationMode.GREEDY_SEARCH)
  • [C语言入门] 结构体
  • 接口自动化学习二：session自动管理cookie
  • Maven+Spring实现后端开发
  • Pytorch中预置数据集的加载方式
  • 数据结构学习
  • 大模型学习二：DeepSeek R1+蒸馏模型组本地部署与调用
  • 【MyBatis】深入解析 MyBatis：关于注解和 XML 的 MyBatis 开发方案下字段名不一致的的查询映射解决方案
  • 操作系统（二）：实时系统介绍与实例分析
  • python match case语法
  • Vue3 Pinia Store 新建store示例、使用store示例
  • 【大模型】SpringBoot整合LangChain4j实现RAG检索实战详解
  • ros2--gazebo--launch
  • 【gdutthesis模板】章节标题有英文解决方案
  • Tmux 核心操作速查指南
  • 【c++深入系列】：类与对象详解（中）