【从零实现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;};}
}

👥总结

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

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

请添加图片描述

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/web/74420.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

AI与玩具结合的可行性分析

文章目录 一、市场需求&#xff1a;教育与陪伴的双重驱动&#xff08;一&#xff09;教育需求&#xff08;二&#xff09;情感陪伴需求&#xff08;三&#xff09;消费升级 二、技术发展&#xff1a;赋能玩具智能化&#xff08;一&#xff09;AI技术的成熟&#xff08;二&#…

基于 RK3588 的 YOLO 多线程推理多级硬件加速引擎框架设计(代码框架和实现细节)

一、前言 接续上一篇文章&#xff0c;这个部分主要分析代码框架的实现细节和设计理念。 基于RK3588的YOLO多线程推理多级硬件加速引擎框架设计&#xff08;项目总览和加速效果&#xff09;-CSDN博客https://blog.csdn.net/plmm__/article/details/146542002?spm1001.2014.300…

LeetCode Hot100 刷题笔记(7)—— 贪心

目录 前言 一、贪心 1. 买卖股票的最佳时机 2. 跳跃游戏 3. 跳跃游戏 II 4. 划分字母区间 前言 一、贪心&#xff1a;买卖股票的最佳时机&#xff0c;跳跃游戏&#xff0c;跳跃游戏 II&#xff0c;划分字母区间。 一、贪心 1. 买卖股票的最佳时机 原题链接&#xff1a;121. …

SQL语句的训练

DELECT FROM 蜀国 WHEHE name 刘玄德 AND 创业进度<0.5 AND 存活状态 true&#xff1b; 基础的sql语句 SELECT >选择列FROM >确认数据源JOIN >联合操作WHERE >筛选数据GROUP BY >分组 HAVING >过滤分组的数据DISTINCT >去重ORDEY BY > 排序…

汽车 HMI 设计的发展趋势与设计要点

一、汽车HMI设计的发展历程与现状 汽车人机交互界面&#xff08;HMI&#xff09;设计经历了从简单到复杂、从单一到多元的演变过程。2012年以前&#xff0c;汽车HMI主要依赖物理按键进行操作&#xff0c;交互方式较为单一。随着特斯拉Model S的推出&#xff0c;触控屏逐渐成为…

基于51单片机的模拟条形码识别系统proteus仿真

地址&#xff1a; https://pan.baidu.com/s/1AtAry19X3BgavLqXcM4scg 提取码&#xff1a;1234 仿真图&#xff1a; 芯片/模块的特点&#xff1a; AT89C52/AT89C51简介&#xff1a; AT89C51 是一款常用的 8 位单片机&#xff0c;由 Atmel 公司&#xff08;现已被 Microchip 收…

CD22.【C++ Dev】类和对象(13) 流提取运算符的重载和const成员

目录 1.流提取运算符>>的重载 知识回顾 重载方法 operator<<格式 operator>>格式 使用cin对日期类对象写入数据 如果想指定格式输入 方法1:getchar() 方法2:使用临时变量接收字符 完善operator>>代码(修bug) 2.类中的权限问题(const成员) …

Spring 核心技术解析【纯干货版】- XIX:Spring 日志模块 Spring-Jcl 模块精讲

在现代 Java 开发中&#xff0c;日志是调试、监控和维护应用程序的重要工具。Spring 作为企业级框架&#xff0c;提供了 Spring-Jcl 作为日志抽象层&#xff0c;使开发者可以灵活切换不同的日志实现&#xff0c;而无需修改业务代码。本篇文章将深入解析 Spring-Jcl 模块&#x…

Hadoop集群---运维管理和技巧

一. daemon 守护进程管理 1. NameNode守护进程管理 hadoop-daemon.sh start namenode 2. DataNode守护进程管理 hadoop-daemon.sh start datanode 3. ResourceManager守护进程管理 yarn-daemon.sh start resourcemanager 4. NodeManager守护进程管理 yarn-daemon.sh st…

ngx_log_init

定义在 src\core\ngx_log.c ngx_log_t * ngx_log_init(u_char *prefix, u_char *error_log) {u_char *p, *name;size_t nlen, plen;ngx_log.file &ngx_log_file;ngx_log.log_level NGX_LOG_NOTICE;if (error_log NULL) {error_log (u_char *) NGX_ERROR_LOG_PATH;}…

网络华为HCIA+HCIP 策略路由,双点双向

目录 路由策略&#xff0c;策略路由 策略路由优势 策略路由分类 接口策略路由 双点双向 双点双向路由引入特点: 联系 路由回灌和环路问题 路由策略&#xff0c;策略路由 路由策略:是对路由条目进行控制&#xff0c;通过控制路由条目影响报文的转发路径&#xff0c;即路…

水下成像机理分析

一般情况下, 水下环境泛指浸入到人工水体 (如水库、人工湖等)或自然水体(如海洋、河流、湖 泊、含水层等)中的区域。在水下环境中所拍摄 的图像由于普遍受到光照、波长、水中悬浮颗粒物 等因素的影响&#xff0c;导致生成的水下图像出现模糊、退 化、偏色等现象&#xff0c;图像…

MySQL的数据目录以及日志

1.MySQL数据目录 MySQL服务器的管理信息、业务数据、⽇志⽂件、磁盘缓冲⽂件默认存储在数据⽬录下.数据目录保存了我们用户的信息,以及我们创建的数据库和表的数据.维护了日志文件等.mysqld主要操作的就是我们的数据目录. 如何查看数据目录: ll /var/lib/mysql#ll 是查看指令 …

论文阅读:Dual Anchor Graph Fuzzy Clustering for Multiview Data

论文地址:Dual Anchor Graph Fuzzy Clustering for Multiview Data | IEEE Journals & Magazine | IEEE Xplore 代码地址&#xff1a;https://github.com/BBKing49/DAG_FC 摘要 多视角锚图聚类近年来成为一个重要的研究领域&#xff0c;催生了多个高效的方法。然而&#…

32f4,串口1,usart.c.h2025

usart.c #include "sys.h" #include "usart.h" #include "led.h" // #include "stdlib.h" #include "stdarg.h" #include "stdio.h" //加入以下代码,支持printf函数,而不需要选择use MicroLIB #if 1#pragma…

C语言:一组位操作宏

解析协议时&#xff0c;取得位域的值是一种常见操作&#xff0c;这些宏可以辅助我们工作。 /* ** 将x的第n位置1 ** ** x 0x00000000 ** BIT_SET(x, 7) 0x00000080 */ #define BIT_SET(x, n) ((x) | (1 << (n)))/* ** 将x的第n位置为0 ** ** x 0x00000080 ** …

记一个使用BigDecimal所有类型变为整数的问题

场景 通过 Excel 导入数据&#xff0c;数据中包含金额。数据库类型 decimal(18, 6) 问题 Excel 导入后所有的金额列都被四舍五入。经过测试&#xff0c;只有数据有整数时所有数据才会被四舍五入&#xff0c;全部为浮点类型没有问题。 解决 强制设置小数位数 // RoundingM…

nodejs、socket.io、express + 实时线上聊天系统(自用笔记)

留个链接给自己参考用&#xff1a; socket.io官方文档&#xff1a;介绍 | Socket.IO nodejs基础语法&#xff1a;大前端技能讲解&#xff1a;NodeJS、Npm、Es6、Webpack_nodejs webpack-CSDN博客 socket.io教学&#xff1a;半小时学会socket.io【中英字幕】Learn Socket.Io …

配置网络编辑器

网络断开的原因 1.由于网络未连接的情况 解决方法 方法1&#xff1a;检查网卡配置 cd /etc/syscongfig/network_scripts vi ifcfg_ens31 方法2&#xff1a;打开虚拟机编辑--- 虚拟网络编辑器 查看ip地址是否在可用的网段范围内 修改后重启网络 systemctl restart netwo…

vscode代码片段的设置与使用

在 Visual Studio Code (VS Code) 中&#xff0c;可以通过自定义**代码片段&#xff08;Snippets&#xff09;**快速插入常用代码模板。以下是详细设置步骤&#xff1a; 步骤 1&#xff1a;打开代码片段设置 按下快捷键 Ctrl Shift P&#xff08;Windows/Linux&#xff09;或…