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

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

backend 2025/8/11 18:57:02

 

目录

创建 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.xdnf.cn/news/17069.html

相关文章：

  • 【计算机网络 | 第3篇】物理媒介
  • 【计算机网络】王道考研笔记整理（3）数据链路层
  • 12、Docker Compose 安装 Redis
  • Baumer相机如何通过YoloV8深度学习模型实现农作物水稻病虫害的检测识别（C#代码UI界面版）
  • PHP官方及第三方下载地址全指南（2025最新版）
  • 芯片封装（DIP、SOP、QFP、QFN、BGA、LGA、PGA）
  • 加载量化模型
  • 第十八天：C++进制之间的转换
  • React 表单处理：移动端输入场景下的卡顿问题与防抖优化方案
  • 【文献分享】Machine learning models提供数据和代码
  • 当前就业形势下，软件测试工程师职业发展与自我提升的必要性
  • JSON巴巴 - 专业JSON格式化工具：让任何JSON都能完美格式化
  • 支持多网络协议的测试工具（postman被无视版）
  • Enhancing Long Video Question Answering with Scene-Localized Frame Grouping
  • 从“T+1”到“T+0”：基于SQL构建MES到数据仓库的数据采集通道
  • SassSCSS：让CSS拥有超能力的预处理器
  • LVS-DR模式高性能负载均衡实战
  • C语言：栈的实现和剖析
  • css怪异模式（Quirks Mode）和标准模式（Standards Mode）最明显的区别
  • 【Java String】类深度解析：从原理到高效使用技巧
  • 软件架构：系统结构的顶层设计与战略约束
  • webrtc弱网-OveruseFrameDetector源码分析与算法原理
  • C++ 类和对象(1)
  • 【qt5_study】1.Hello world
  • SpringCloud学习------Hystrix详解
  • 奇偶校验码原理与FPGA实现
  • ubuntu自动重启BUG排查指南
  • Android 性能基准测试(Benchmark)完全指南：专业方法与最佳实践
  • 【RK3576】【Android14】Uboot下fastboot命令支持
  • 磁悬浮转子振动控制：主动电磁力如何成为高速旋转的“振动克星”