连接管理模块和服务器模块

Ne0inhk

— 15 min read
连接管理模块和服务器模块

1. 封装连接管理类

向用户提供一个用于实现网络通信的 Connection 对象，从其内部可创建出粒度更轻的Channel 对象，用于与客户端进行网络通信。

  1. 成员信息：
  • 连接关联的信道管理句柄（实现信道的增删查）
  • 连接关联的实际用于通信的 muduo::net::Connection 连接
  • protobuf 协议处理的句柄（ProtobufCodec 对象）
  • 消费者管理句柄
  • 虚拟机句柄
  • 异步工作线程池句柄
  1. 连接操作：
  • 提供创建 Channel 信道的操作
  • 提供删除 Channel 信道的操作
  1. 连接管理：
  • 连接的增删查

为什么需要这些成员和操作？

  1. 信道管理句柄：因为AMQP协议允许在一个连接上创建多个信道，每个信道可以独立进行操作（如声明队列、发布消息等）。所以连接管理模块需要能够管理这些信道，包括创建、删除和查找。
  2. muduo::net::Connection：这是实际进行网络通信的对象，连接管理模块需要持有它以便进行数据的发送和接收，同时也需要监听连接的事件（如断开、消息到达等）。
  3. ProtobufCodec：因为我们的通信协议是使用protobuf序列化的，所以需要这个编解码器来解析和封装消息。
  4. 消费者管理句柄：消费者是消息队列的重要概念，连接管理模块需要消费者管理句柄来管理消费者（因为消费者是建立在连接上的，虽然具体操作在信道，但消费者资源属于连接）。在Channel类中，我们看到了消费者管理器的使用，所以连接管理模块需要持有消费者管理器，并传递给每个信道。
  5. 虚拟机句柄：虚拟机是RabbitMQ中资源隔离的单位，连接必须属于一个虚拟机。连接管理模块需要虚拟机句柄来访问虚拟机中的交换机、队列等资源。
  6. 异步工作线程池句柄：为了不阻塞网络线程，一些耗时的操作（如消息的消费）需要放到线程池中执行。因此，连接管理模块需要持有线程池的句柄，并传递给信道使用。

操作方面：
创建和删除信道是AMQP协议的基本操作，因为客户端可以通过打开和关闭信道来在同一个连接上实现多路复用。

连接管理（增删查）则是对连接本身的管理，当客户端建立连接时，服务器需要创建一个连接对象，并管理起来，以便在连接断开时清理资源。

classConnection{public:using ptr = std::shared_ptr<Connection>;Connection(const VirtualHost::ptr &host,const ConsumerManager::ptr &cmp,const ProtobufCodecPtr &codec,const muduo::net::TcpConnectionPtr &conn,const threadpool::ptr &pool):_conn(conn),_codec(codec),_cmp(cmp),_host(host),_pool(pool),_channels(std::make_shared<ChannelManager>()){}voidopenChannel(const openChannelRequestPtr& req){//1. 判断信道ID是否重复,创建信道bool ret = _channels->openChannel(req->rid(), _host, _cmp, _codec, _conn, _pool);if(ret ==false){DLOG("创建信道的时候，信道ID重复了");basicResponse(false, req->rid(), req->cid());return;}DLOG("%s 信道创建成功！", req->cid().c_str());//2. 给客户端进行回复basicResponse(true, req->rid(), req->cid());}voidcloseChannel(const closeChannelRequestPtr& req){ _channels->closeChannel(req->cid());basicResponse(true, req->rid(), req->cid());} Channel::ptr getChannel(const std::string &cid){return _channels->getChannel(cid);}private:voidbasicResponse(bool ok,const std::string &rid,const std::string &cid){ basicCommonResponse resp; resp.set_rid(rid); resp.set_cid(cid); resp.set_ok(ok); _codec->send(_conn, resp);}private: muduo::net::TcpConnectionPtr _conn; ProtobufCodecPtr _codec; ConsumerManager::ptr _cmp; VirtualHost::ptr _host; threadpool::ptr _pool; ChannelManager::ptr _channels;};

2. 封装对外管理的类

classConnectionManager{public:using ptr = std::shared_ptr<ConnectionManager>;ConnectionManager(){}voidnewConnection(const VirtualHost::ptr &host,const ConsumerManager::ptr &cmp,const ProtobufCodecPtr &codec,const muduo::net::TcpConnectionPtr &conn,const threadpool::ptr &pool){ std::unique_lock<std::mutex>lock(_mutex);auto it = _conns.find(conn);if(it != _conns.end())return; Connection::ptr self_conn = std::make_shared<Connection>(host, cmp, codec, conn, pool); _conns.insert(std::make_pair(conn, self_conn));}voiddelConnection(const muduo::net::TcpConnectionPtr &conn){ std::unique_lock<std::mutex>lock(_mutex); _conns.erase(conn);} Connection::ptr getConnection(const muduo::net::TcpConnectionPtr &conn){ std::unique_lock<std::mutex>lock(_mutex);auto it = _conns.find(conn);if(it == _conns.end()){returnConnection::ptr();}return it->second;}private: std::mutex _mutex; std::unordered_map<muduo::net::TcpConnectionPtr, Connection::ptr> _conns;};

服务器通过对外管理的类来实现对连接的增删查操作

3. 服务器管理

这里我们考虑客户端和服务器之间的通信方式。回顾 MQ 的交互模型：

其中生产者和消费者都是客户端， 它们都需要通过网络和 Broker Server 进行通信。具体通信的过程我们使用 Muduo 库来实现， 使用 TCP 作为通信的底层协议， 同时在这个基础上自定义应用层协议， 完成客户端对服务器功能的远端调用。 我们要实现的远端调用接口包括：

  • 创建 channel
  • 关闭 channel
  • 创建 exchange
  • 删除 exchange
  • 创建 queue
  • 删除 queue
  • 创建 binding
  • 删除 binding
  • 发送 message
  • 订阅 message
  • 发送 ack
  • 返回 message (服务器 -> 客户端)

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

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

关系图提示：

图中显示了多个回调函数，如onConnection、onUnknownMessage、onChannelOpen、onExchangeDeclare、onQueueDelete等，这些是_dispatcher中注册的回调函数，用于处理不同类型的消息。

流程分析：

  1. 服务器启动：

使用Muduo的TcpServer，设置_baseloop为主事件循环，监听指定端口。

  1. 连接建立：

当有新连接建立时，会调用onConnection回调。在这个回调中，可能会创建一个Connection对象，并加入到_connections中进行管理。

  1. 消息接收：

当有数据到达时，TcpServer会从socket读取数据，并调用设置的消息回调。这里，我们使用_codec进行解码。

  1. 消息解码：

_codec是protobuf编解码器，它按照我们定义的应用层协议（例如，可能有长度字段+protobuf数据）进行解码，得到完整的protobuf消息。

  1. 消息分发：

解码后的protobuf消息会被送到dispatcher。dispatcher根据消息的类型（protobuf的描述符中的全名）查找对应的回调函数。

  1. 消息处理：

根据消息类型，调用注册的回调函数。例如，如果是声明交换机的消息，则调用onExchangeDeclare；如果是声明队列的消息，则调用onQueueDeclare；如果是发布消息，则调用onBasicPublish等。

  1. 处理函数内部：

在回调函数中，会进行相应的业务处理。例如，声明交换机就会在virtual_host中创建交换机；发布消息就会将消息通过交换机路由到队列，然后可能通过threadpool异步推送给消费者。

  1. 消费者管理：

消费者管理由_consumer句柄负责。当有消费者订阅队列时，会记录消费者信息，并在有消息时通过线程池将消息推送给消费者。

  1. 连接管理：

_connections管理所有连接，当连接断开时，需要清理该连接相关的资源，比如该连接上的信道、消费者等。

  1. 虚拟主机：

所有交换机、队列、绑定、消息的持久化等都在_virtual_host中管理，它是整个消息存储和转发的核心。

  1. 异步处理：

使用_threadpool进行异步消息推送，避免阻塞网络IO线程。

  1. 未知消息处理：

如果收到未知类型的消息，会调用onUnknownMessage进行处理，可能会返回错误。

具体到关系图中的回调函数，它们被注册到_dispatcher的callbackMap中，当对应类型的消息到达时，就会调用。

总结：

整个服务器是一个基于事件驱动、异步处理的消息中间件服务器。它通过protobuf定义消息格式，利用Muduo处理网络IO，利用线程池处理耗时的消息推送，通过虚拟主机管理所有的消息数据，并通过连接管理和消费者管理来维护客户端的状态。

4. 服务器完整代码

#ifndef__M_BROKER_H__#define__M_BROKER_H__#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"../mqcommon/threadpool.hpp"#include"../mqcommon/msg.pb.h"#include"../mqcommon/proto.pb.h"#include"../mqcommon/logger.hpp"#include"connection.hpp"#include"consumer.hpp"#include"host.hpp"namespace rabbitmq {#defineDBFILE"/meta.db"#defineHOSTNAME"MyVirtualHost"classServer{public:typedef std::shared_ptr<google::protobuf::Message> MessagePtr;Server(int port,const std::string &basedir):_server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port),"Server", muduo::net::TcpServer::kReusePort),_dispatcher(std::bind(&Server::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))),_virtual_host(std::make_shared<VirtualHost>(HOSTNAME, basedir, basedir + DBFILE)),_consumer_manager(std::make_shared<ConsumerManager>()),_connection_manager(std::make_shared<ConnectionManager>()),_threadpool(std::make_shared<threadpool>()){//针对历史消息中的所有队列，还需要初始化队列的消费者管理结构 QueueMap qm = _virtual_host->allQueues();for(auto& q : qm){ _consumer_manager->initQueueConsumer(q.first);}// 注册业务请求处理函数 _dispatcher.registerMessageCallback<rabbitmq::openChannelRequest>(std::bind(&Server::onOpenChannel,this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)); _dispatcher.registerMessageCallback<rabbitmq::closeChannelRequest>(std::bind(&Server::onCloseChannel,this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)); _dispatcher.registerMessageCallback<rabbitmq::declareExchangeRequest>(std::bind(&Server::onDeclareExchange,this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)); _dispatcher.registerMessageCallback<rabbitmq::deleteExchangeRequest>(std::bind(&Server::onDeleteExchange,this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)); _dispatcher.registerMessageCallback<rabbitmq::declareQueueRequest>(std::bind(&Server::onDeclareQueue,this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)); _dispatcher.registerMessageCallback<rabbitmq::deleteQueueRequest>(std::bind(&Server::onDeleteQueue,this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)); _dispatcher.registerMessageCallback<rabbitmq::queueBindRequest>(std::bind(&Server::onQueueBind,this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)); _dispatcher.registerMessageCallback<rabbitmq::queueUnBindRequest>(std::bind(&Server::onQueueUnBind,this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)); _dispatcher.registerMessageCallback<rabbitmq::basicPublishRequest>(std::bind(&Server::onBasicPublish,this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)); _dispatcher.registerMessageCallback<rabbitmq::basicAckRequest>(std::bind(&Server::onBasicAck,this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)); _dispatcher.registerMessageCallback<rabbitmq::basicConsumeRequest>(std::bind(&Server::onBasicConsume,this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)); _dispatcher.registerMessageCallback<rabbitmq::basicCancelRequest>(std::bind(&Server::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(&Server::onConnection,this, std::placeholders::_1));}voidstart(){ _server.start(); _baseloop.loop();}private://打开信道voidonOpenChannel(const muduo::net::TcpConnectionPtr &conn,const openChannelRequestPtr &message, muduo::Timestamp){ Connection::ptr mconn = _connection_manager->getConnection(conn);if(mconn.get()==nullptr){DLOG("打开信道时, 没有找到连接对应的Connection对象!"); conn->shutdown();return;} mconn->openChannel(message);}//关闭信道voidonCloseChannel(const muduo::net::TcpConnectionPtr &conn,const closeChannelRequestPtr &message, muduo::Timestamp){ Connection::ptr mconn = _connection_manager->getConnection(conn);if(mconn.get()==nullptr){DLOG("关闭信道时, 没有找到连接对应的Connection对象!"); conn->shutdown();return;} mconn->closeChannel(message);}//声明交换机voidonDeclareExchange(const muduo::net::TcpConnectionPtr &conn,const declareExchangeRequestPtr &message, muduo::Timestamp){ Connection::ptr mconn = _connection_manager->getConnection(conn);if(mconn.get()==nullptr){DLOG("声明交换机时, 没有找到连接对应的Connection对象!"); conn->shutdown();return;} Channel::ptr cp = mconn->getChannel(message->cid());if(cp.get()==nullptr){DLOG("声明交换机时, 没有找到信道!");return;} cp->declareExchange(message);}//删除交换机voidonDeleteExchange(const muduo::net::TcpConnectionPtr &conn,const deleteExchangeRequestPtr &message, muduo::Timestamp){ Connection::ptr mconn = _connection_manager->getConnection(conn);if(mconn.get()==nullptr){DLOG("删除交换机时, 没有找到连接对应的Connection对象!"); conn->shutdown();return;} Channel::ptr cp = mconn->getChannel(message->cid());if(cp.get()==nullptr){DLOG("删除交换机时, 没有找到信道!");return;} cp->deleteExchange(message);}//声明队列voidonDeclareQueue(const muduo::net::TcpConnectionPtr &conn,const declareQueueRequestPtr &message, muduo::Timestamp){ Connection::ptr mconn = _connection_manager->getConnection(conn);if(mconn.get()==nullptr){DLOG("声明队列时, 没有找到连接对应的Connection对象!"); conn->shutdown();return;} Channel::ptr cp = mconn->getChannel(message->cid());if(cp.get()==nullptr){DLOG("声明队列时, 没有找到信道!");return;} cp->declareQueue(message);}//删除队列voidonDeleteQueue(const muduo::net::TcpConnectionPtr &conn,const deleteQueueRequestPtr &message, muduo::Timestamp){ Connection::ptr mconn = _connection_manager->getConnection(conn);if(mconn.get()==nullptr){DLOG("删除队列时, 没有找到连接对应的Connection对象!"); conn->shutdown();return;} Channel::ptr cp = mconn->getChannel(message->cid());if(cp.get()==nullptr){DLOG("删除队列时, 没有找到信道!");return;} cp->deleteQueue(message);}//队列绑定voidonQueueBind(const muduo::net::TcpConnectionPtr &conn,const queueBindRequestPtr &message, muduo::Timestamp){ Connection::ptr mconn = _connection_manager->getConnection(conn);if(mconn.get()==nullptr){DLOG("队列绑定时, 没有找到连接对应的Connection对象!"); conn->shutdown();return;} Channel::ptr cp = mconn->getChannel(message->cid());if(cp.get()==nullptr){DLOG("队列绑定时, 没有找到信道!");return;} cp->queueBind(message);}//队列解绑voidonQueueUnBind(const muduo::net::TcpConnectionPtr &conn,const queueUnBindRequestPtr &message, muduo::Timestamp){ Connection::ptr mconn = _connection_manager->getConnection(conn);if(mconn.get()==nullptr){DLOG("队列解绑时, 没有找到连接对应的Connection对象!"); conn->shutdown();return;} Channel::ptr cp = mconn->getChannel(message->cid());if(cp.get()==nullptr){DLOG("队列解绑时, 没有找到信道!");return;} cp->queueUnBind(message);}//消息发布voidonBasicPublish(const muduo::net::TcpConnectionPtr &conn,const basicPublishRequestPtr &message, muduo::Timestamp){ Connection::ptr mconn = _connection_manager->getConnection(conn);if(mconn.get()==nullptr){DLOG("消息发布时, 没有找到连接对应的Connection对象!"); conn->shutdown();return;} Channel::ptr cp = mconn->getChannel(message->cid());if(cp.get()==nullptr){DLOG("消息发布时, 没有找到信道!");return;} cp->basicPublish(message);}//消息确认voidonBasicAck(const muduo::net::TcpConnectionPtr &conn,const basicAckRequestPtr &message, muduo::Timestamp){ Connection::ptr mconn = _connection_manager->getConnection(conn);if(mconn.get()==nullptr){DLOG("消息确认时, 没有找到连接对应的Connection对象!"); conn->shutdown();return;} Channel::ptr cp = mconn->getChannel(message->cid());if(cp.get()==nullptr){DLOG("消息确认时, 没有找到信道!");return;} cp->basicAck(message);}//队列消息订阅voidonBasicConsume(const muduo::net::TcpConnectionPtr &conn,const basicConsumeRequestPtr &message, muduo::Timestamp){ Connection::ptr mconn = _connection_manager->getConnection(conn);if(mconn.get()==nullptr){DLOG("队列消息订阅时, 没有找到连接对应的Connection对象!"); conn->shutdown();return;} Channel::ptr cp = mconn->getChannel(message->cid());if(cp.get()==nullptr){DLOG("队列消息订阅时, 没有找到信道!");return;} cp->basicConsume(message);}//队列消息取消订阅voidonBasicCancel(const muduo::net::TcpConnectionPtr &conn,const basicCancelRequestPtr &message, muduo::Timestamp){ Connection::ptr mconn = _connection_manager->getConnection(conn);if(mconn.get()==nullptr){DLOG("队列消息取消订阅时, 没有找到连接对应的Connection对象!"); conn->shutdown();return;} Channel::ptr cp = mconn->getChannel(message->cid());if(cp.get()==nullptr){DLOG("队列消息取消订阅时, 没有找到信道!");return;} cp->basicCancel(message);}voidonUnknownMessage(const muduo::net::TcpConnectionPtr &conn,const MessagePtr &message, muduo::Timestamp){ LOG_INFO <<"onUnknownMessage: "<< message->GetTypeName(); conn->shutdown();}voidonConnection(const muduo::net::TcpConnectionPtr &conn){if(conn->connected()){ _connection_manager->newConnection(_virtual_host, _consumer_manager, _codec, conn, _threadpool);}else{ _connection_manager->delConnection(conn);}}private: muduo::net::EventLoop _baseloop; muduo::net::TcpServer _server;// 服务器对象 ProtobufDispatcher _dispatcher;// 请求分发器对象--要向其中注册请求处理函数 ProtobufCodecPtr _codec;// protobuf协议处理器--针对收到的请求数据进行protobuf协议处理 VirtualHost::ptr _virtual_host; ConsumerManager::ptr _consumer_manager; ConnectionManager::ptr _connection_manager; threadpool::ptr _threadpool;};}#endif

Read more

【MYSQL】MYSQL学习的一大重点：MYSQL库的操作

【MYSQL】MYSQL学习的一大重点：MYSQL库的操作

🎬 个人主页：艾莉丝努力练剑 ❄专栏传送门：《C语言》《数据结构与算法》《C/C++干货分享&学习过程记录》 《Linux操作系统编程详解》《笔试/面试常见算法：从基础到进阶》《Python干货分享》 ⭐️为天地立心，为生民立命，为往圣继绝学，为万世开太平 🎬 艾莉丝的简介： 文章目录 * 0 ~> 实际场景：创建和删除数据库 * 0.1 创建方式1 * 0.2 创建方式2 * 0.3 创建方式3 * 1 ~> 数据库的编码集 * 1.1 目前整个数据库支持的字符集 * 1.2 目前整个数据库支持的字符集 * 1.3 UTF-8需要设置配置文件 * 1.4 MySQL 中与字符集排序规则（

By Ne0inhk
Spring Boot + jQuery 前后端分离图书管理系统：从接口设计到问题排查

Spring Boot + jQuery 前后端分离图书管理系统：从接口设计到问题排查

图书管理系统 1.1 准备前端代码 在本地想要的可以去我的gitee中下载 library 的相关前端代码 1.2 约定前后端交互接口 需求分析 图书管理系统是⼀个相对较大一点的案例,咱们先实现其中的⼀部分功能. 用户登录 1. 登录接口 2. 图书列表展示 字段说明: 字段说明id图书 IDbookName图书名称author作者count数量price定价publish图书出版社status图书状态 1 - 可借阅 其他 - 不可借阅statusCN图书状态中文含义 3.4.3 服务器代码 创建图书类 BookInfo @Data public class BookInfo { //图书ID private Integer id; //书名 private String bookName; //作者 private String

By Ne0inhk
深入剖析Spring框架：架构、缺陷与演进之路

深入剖析Spring框架：架构、缺陷与演进之路

深入剖析Spring框架：架构、缺陷与演进之路 * 引言：Spring的辉煌与挑战 * 一、Spring源码架构分析 * 1.1 整体架构：模块化的艺术 * 核心容器（Core Container） * 1.2 IoC容器：Spring的心脏 * 1.3 AOP实现：优雅的横切关注点解决方案 * 二、Spring的缺陷与不足 * 2.1 性能瓶颈：反射的代价 * 2.2 配置复杂性：灵活性的双刃剑 * 2.3 启动时间：云原生时代的痛点 * 2.4 响应式编程的局限性 * 三、改进Spring的方案 * 3.1 编译时增强：GraalVM与Spring Native * 3.2 模块化精简：面向云原生的瘦身

By Ne0inhk