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

C++ - 仿 RabbitMQ 实现消息队列--服务器模块实现

news 2025/8/7 6:57:23

 

目录

创建 MQBrokerServer 

       服务器模块我们借助 Muduo 网络库来实现。        

  • _server：Muduo 库提供的一个通用 TCP 服务器， 我们可以封装这个服务器进行TCP 通信。
  • _baseloop：主事件循环器， 用于响应 IO 事件和定时器事件，主 loop 主要是为了响应监听描述符的 IO 事件。
  • _codec: 一个 protobuf 编解码器， 我们在 TCP 服务器上设计了一层应用层协议，这个编解码器主要就是负责实现应用层协议的解析和封装。
  • _dispatcher：一个消息分发器， 当 Socket 接收到一个报文消息后， 我们需要按照消息的类型， 即上面提到的 typeName 进行消息分发， 会不不同类型的消息分发相对应的的处理函数中。
  • _consumer: 服务器中的消费者信息管理句柄。
  • _threadpool: 异步工作线程池，主要用于队列消息的推送工作。
  • _connections: 连接管理句柄，管理当前服务器上的所有已经建立的通信连接。
  • _virtual_host：服务器持有的虚拟主机。 队列、交换机 、绑定、消息等数据都是通过虚拟主机管理。

创建 MQBrokerServer 

        BrokerServer 模块是对整体服务器所有模块的整合，接收客户端的请求，并提供服务。基于前边实现的简单的翻译服务器代码，进行改造，只需要实现服务器内部提供服务的各个业务接口即可。
        在各个业务处理函数中，也比较简单，创建信道后，每次请求过来后，找到请求对应的信道句柄，通过句柄调用前边封装好的处理接口进行请求处理，最终返回处理结果。

#pragma once
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"#include "connection.hpp"
#include "consumer.hpp"
#include "vhost.hpp"
#include "../mqcommon/log.hpp"
#include "../mqcommon/msg.pb.h"
#include "../mqcommon/proto.pb.h"
#include "../mqcommon/threadpool.hpp"#include <iostream>
#include <unistd.h>
#include <unordered_map>namespace jiuqi
{#define DBFILE "/meta.db"#define VHOST "vhost"class BrokerServer{public:using MessagePtr = std::shared_ptr<google::protobuf::Message>;BrokerServer(int port, const std::string &basicDir): _server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port),"server", muduo::net::TcpServer::kReusePort),_dispatcher(std::bind(&BrokerServer::onUnknownMessage, this,std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)),_codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher,std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),_vhost(std::make_shared<VirtualHost>(VHOST, basicDir, basicDir + DBFILE)),_consumer_manager(std::make_shared<ConsumerManager>()),_connection_manager(std::make_shared<ConnectionManager>()),_threadpool(std::make_shared<ThreadPool>()){QueueMap qm = _vhost->allqueue();for (auto &queue : qm){_consumer_manager->initQueueConsumer(queue.first);}// 注册请求处理函数_dispatcher.registerMessageCallback<jiuqi::openChannelRequest>(std::bind(&BrokerServer::onOpenChannel, this,std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_dispatcher.registerMessageCallback<jiuqi::closeChannelRequest>(std::bind(&BrokerServer::onCloseChannel, this,std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_dispatcher.registerMessageCallback<jiuqi::declareExchangeRequest>(std::bind(&BrokerServer::onDeclareExchange, this,std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_dispatcher.registerMessageCallback<jiuqi::deleteExchangeRequest>(std::bind(&BrokerServer::onDeleteExchange, this,std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_dispatcher.registerMessageCallback<jiuqi::declareQueueRequest>(std::bind(&BrokerServer::onDeclareQueue, this,std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_dispatcher.registerMessageCallback<jiuqi::deleteQueueRequest>(std::bind(&BrokerServer::onDeleteQueue, this,std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_dispatcher.registerMessageCallback<jiuqi::queueBindRequest>(std::bind(&BrokerServer::onQueueBind, this,std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_dispatcher.registerMessageCallback<jiuqi::queueUnbindRequest>(std::bind(&BrokerServer::onQueueUnbind, this,std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_dispatcher.registerMessageCallback<jiuqi::basicPublishRequest>(std::bind(&BrokerServer::onBasicPublish, this,std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_dispatcher.registerMessageCallback<jiuqi::basicAckRequest>(std::bind(&BrokerServer::onBasicAck, this,std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_dispatcher.registerMessageCallback<jiuqi::basicConsumeRequest>(std::bind(&BrokerServer::onBasicConsume, this,std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_dispatcher.registerMessageCallback<jiuqi::basicCancelRequest>(std::bind(&BrokerServer::onBasicCancel, this,std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(),std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_server.setConnectionCallback(std::bind(&BrokerServer::onConnection, this, std::placeholders::_1));}void start(){_server.start();_baseloop.loop();}private:// 打开信道void onOpenChannel(const muduo::net::TcpConnectionPtr &conn,const openChannelRequestPtr &message,muduo::Timestamp){Connection::ptr mconn = _connection_manager->getConnection(conn);if (mconn == nullptr) {DEBUG("打开信道时,没有找到对应的连接对象");conn->shutdown();return;}mconn->openChannel(message);}// 关闭信道void onCloseChannel(const muduo::net::TcpConnectionPtr &conn,const closeChannelRequestPtr &message,muduo::Timestamp){Connection::ptr mconn = _connection_manager->getConnection(conn);if (mconn == nullptr) {DEBUG("关闭信道时,没有找到对应的连接对象");conn->shutdown();return;}mconn->closeChannel(message);}// 声明交换机void onDeclareExchange(const muduo::net::TcpConnectionPtr &conn,const declareExchangeRequestPtr &message,muduo::Timestamp){Connection::ptr mconn = _connection_manager->getConnection(conn);if (mconn == nullptr) {DEBUG("声明交换机时,没有找到对应的连接对象");conn->shutdown();return;}Channel::ptr channel = mconn->getChannel(message->cid());if (channel == nullptr){DEBUG("声明交换机时,没有找到对应的信道");return;}channel->declareExchange(message);}// 删除交换机void onDeleteExchange(const muduo::net::TcpConnectionPtr &conn,const deleteExchangeRequestPtr &message,muduo::Timestamp){Connection::ptr mconn = _connection_manager->getConnection(conn);if (mconn == nullptr) {DEBUG("删除交换机时,没有找到对应的连接对象");conn->shutdown();return;}Channel::ptr channel = mconn->getChannel(message->cid());if (channel == nullptr){DEBUG("删除交换机时,没有找到对应的信道");return;}channel->deleteExchange(message);           }// 声明队列void onDeclareQueue(const muduo::net::TcpConnectionPtr &conn,const declareQueueRequestPtr &message,muduo::Timestamp){Connection::ptr mconn = _connection_manager->getConnection(conn);if (mconn == nullptr) {DEBUG("声明队列时,没有找到对应的连接对象");conn->shutdown();return;}Channel::ptr channel = mconn->getChannel(message->cid());if (channel == nullptr){DEBUG("声明队列时,没有找到对应的信道");return;}channel->declareQueue(message);            }// 删除队列void onDeleteQueue(const muduo::net::TcpConnectionPtr &conn,const deleteQueueRequestPtr &message,muduo::Timestamp){Connection::ptr mconn = _connection_manager->getConnection(conn);if (mconn == nullptr) {DEBUG("删除队列时,没有找到对应的连接对象");conn->shutdown();return;}Channel::ptr channel = mconn->getChannel(message->cid());if (channel == nullptr){DEBUG("删除队列时,没有找到对应的信道");return;}channel->deleteQueue(message);                  }// 队列绑定void onQueueBind(const muduo::net::TcpConnectionPtr &conn,const queueBindRequestPtr &message,muduo::Timestamp){Connection::ptr mconn = _connection_manager->getConnection(conn);if (mconn == nullptr) {DEBUG("队列绑定时,没有找到对应的连接对象");conn->shutdown();return;}Channel::ptr channel = mconn->getChannel(message->cid());if (channel == nullptr){DEBUG("队列绑定时,没有找到对应的信道");return;}channel->queueBind(message);                 }// 队列解绑void onQueueUnbind(const muduo::net::TcpConnectionPtr &conn,const queueUnbindRequestPtr &message,muduo::Timestamp){Connection::ptr mconn = _connection_manager->getConnection(conn);if (mconn == nullptr) {DEBUG("队列解绑时,没有找到对应的连接对象");conn->shutdown();return;}Channel::ptr channel = mconn->getChannel(message->cid());if (channel == nullptr){DEBUG("队列解绑时,没有找到对应的信道");return;}channel->queueUnbind(message);              }// 消息发布void onBasicPublish(const muduo::net::TcpConnectionPtr &conn,const basicPublishRequestPtr &message,muduo::Timestamp){Connection::ptr mconn = _connection_manager->getConnection(conn);if (mconn == nullptr) {DEBUG("消息发布时,没有找到对应的连接对象");conn->shutdown();return;}Channel::ptr channel = mconn->getChannel(message->cid());if (channel == nullptr){DEBUG("消息发布时,没有找到对应的信道");return;}channel->basicPublish(message);               }// 消息确认void onBasicAck(const muduo::net::TcpConnectionPtr &conn,const basicAckRequestPtr &message,muduo::Timestamp){Connection::ptr mconn = _connection_manager->getConnection(conn);if (mconn == nullptr) {DEBUG("消息确认时,没有找到对应的连接对象");conn->shutdown();return;}Channel::ptr channel = mconn->getChannel(message->cid());if (channel == nullptr){DEBUG("消息确认时,没有找到对应的信道");return;}channel->basicAck(message);                   }// 队列消息订阅void onBasicConsume(const muduo::net::TcpConnectionPtr &conn,const basicConsumeRequestPtr &message,muduo::Timestamp){Connection::ptr mconn = _connection_manager->getConnection(conn);if (mconn == nullptr) {DEBUG("队列消息订阅时,没有找到对应的连接对象");conn->shutdown();return;}Channel::ptr channel = mconn->getChannel(message->cid());if (channel == nullptr){DEBUG("队列消息订阅时,没有找到对应的信道");return;}channel->basicConsumer(message);                  }// 队列消息取消订阅void onBasicCancel(const muduo::net::TcpConnectionPtr &conn,const basicCancelRequestPtr &message,muduo::Timestamp){Connection::ptr mconn = _connection_manager->getConnection(conn);if (mconn == nullptr) {DEBUG("队列消息取消订阅时,没有找到对应的连接对象");conn->shutdown();return;}Channel::ptr channel = mconn->getChannel(message->cid());if (channel == nullptr){DEBUG("队列消息取消订阅时,没有找到对应的信道");return;}channel->basicCancel(message);                 }void onUnknownMessage(const muduo::net::TcpConnectionPtr &conn,const MessagePtr &message,muduo::Timestamp){LOG_INFO << "onUnknownMessage: " << message->GetTypeName();conn->shutdown();}void onConnection(const muduo::net::TcpConnectionPtr &conn){if (conn->connected()){_connection_manager->newConnection(conn, _codec, _consumer_manager, _vhost, _threadpool);LOG_INFO << "连接建立成功";}else{_connection_manager->deleteConnection(conn);LOG_INFO << "连接关闭";}}private:muduo::net::EventLoop _baseloop;muduo::net::TcpServer _server;  // 服务器对象ProtobufDispatcher _dispatcher; // 请求分发器对象--要向其中注册请求处理函数ProtobufCodecPtr _codec;        // protobuf协议处理器--针对收到的请求数据进行prototo协议处理VirtualHost::ptr _vhost;ConsumerManager::ptr _consumer_manager;ConnectionManager::ptr _connection_manager;ThreadPool::ptr _threadpool;};
}

http://www.lryc.cn/news/611695.html

相关文章：

  • Linux网络编程基础-简易TCP服务器框架
  • 服务器——“查询不到显卡驱动，且输入nvidia-smi报错”的解决办法
  • Docker的安装，服务器与客户端之间的通信
  • copy_file_range系统调用及示例
  • 【网络运维】Linux：简单DHCP服务器的部署
  • Profinet转Ethernet IP网关接入五轴车床上下料机械手控制系统的配置实例
  • 03-mysql/redis/apache安装记录
  • 开疆智能ModbusTCP转Profinet网关连接安川YRC1000机器人配置案例
  • PHP官方及第三方下载地址全指南（2025最新版）
  • apache-superset config.py、superset_config.py完整配置项解读
  • SQL的条件查询
  • SQL120 贷款情况
  • CSS高频属性速查指南
  • 基于智能体技术的AIGC源码
  • ABP VNext + SQL Server Temporal Tables：审计与时序数据管理
  • 从 0 到 1：写一个能跑在大体量应用后台的 C++ 协程库
  • 怎么免费建立自己的网站步骤
  • Docker 数据存储路径（解决默认docker路径位置磁盘空间不足的情况）
  • 家庭宽带中的服务器如何被外网访问？
  • RequestBodyAdviceAdapter是什么有什么用
  • [Linux]学习笔记系列 -- [arm][debug]
  • MCP 协议：AI 时代的 “万能转接头”，从 “手动粘贴” 到 “万能接口”：MCP 协议如何重构 AI 工具调用规则？
  • Linux 中 Git 操作大全
  • Go语言 单元测试
  • 鸿蒙app 开发中 全局弹窗类的封装 基于PromptAction
  • LazyLLM教程 | 第3讲：大模型怎么玩：用LazyLLM带你理解调用逻辑与Prompt魔法！
  • AI_提示词Prompt
  • MCP-PromptX AI小说创作使用教程
  • 百度智能云给“数字人”发工牌
  • 纯血鸿蒙（HarmonyOS NEXT）应用开发完全指南