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

1. 封装连接管理类

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

成员信息：

连接关联的信道管理句柄（实现信道的增删查）
连接关联的实际用于通信的 muduo::net::Connection 连接
protobuf 协议处理的句柄（ProtobufCodec 对象）
消费者管理句柄
虚拟机句柄
异步工作线程池句柄

连接操作：

提供创建 Channel 信道的操作
提供删除 Channel 信道的操作

连接管理：

连接的增删查

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

信道管理句柄：因为AMQP协议允许在一个连接上创建多个信道，每个信道可以独立进行操作（如声明队列、发布消息等）。所以连接管理模块需要能够管理这些信道，包括创建、删除和查找。

muduo::net::Connection：这是实际进行网络通信的对象，连接管理模块需要持有它以便进行数据的发送和接收，同时也需要监听连接的事件（如断开、消息到达等）。

ProtobufCodec：因为我们的通信协议是使用protobuf序列化的，所以需要这个编解码器来解析和封装消息。

消费者管理句柄：消费者是消息队列的重要概念，连接管理模块需要消费者管理句柄来管理消费者（因为消费者是建立在连接上的，虽然具体操作在信道，但消费者资源属于连接）。在Channel类中，我们看到了消费者管理器的使用，所以连接管理模块需要持有消费者管理器，并传递给每个信道。

虚拟机句柄：虚拟机是RabbitMQ中资源隔离的单位，连接必须属于一个虚拟机。连接管理模块需要虚拟机句柄来访问虚拟机中的交换机、队列等资源。

异步工作线程池句柄：为了不阻塞网络线程，一些耗时的操作（如消息的消费）需要放到线程池中执行。因此，连接管理模块需要持有线程池的句柄，并传递给信道使用。

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

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

class Connection
{
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>())
{}
void openChannel(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());
}
void closeChannel(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:
void basicResponse(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. 封装对外管理的类

class ConnectionManager
{
public:
using ptr = std::shared_ptr<ConnectionManager>;
ConnectionManager() {}
void newConnection(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));
}
void delConnection(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())
{
return Connection::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中注册的回调函数，用于处理不同类型的消息。

流程分析：

服务器启动：

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

连接建立：

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

消息接收：

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

消息解码：

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

消息分发：

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

消息处理：

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

处理函数内部：

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

消费者管理：

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

连接管理：

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

虚拟主机：

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

异步处理：

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

未知消息处理：

如果收到未知类型的消息，会调用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
{
#define DBFILE "/meta.db"
#define HOSTNAME "MyVirtualHost"
class Server
{
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));
}

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.get() == nullptr)
{
DLOG("打开信道时, 没有找到连接对应的Connection对象!");
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.get() == nullptr)
{
DLOG("关闭信道时, 没有找到连接对应的Connection对象!");
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.get() == nullptr)
{
DLOG("声明交换机时, 没有找到连接对应的Connection对象!");
conn->shutdown();
return;
}
Channel::ptr cp = mconn->getChannel(message->cid());
if(cp.get() == nullptr)
{
DLOG("声明交换机时, 没有找到信道!");
return;
}
cp->declareExchange(message);
}
//删除交换机
void onDeleteExchange(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);
}
//声明队列
void onDeclareQueue(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);
}
//删除队列
void onDeleteQueue(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);
}
//队列绑定
void onQueueBind(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);
}
//队列解绑
void onQueueUnBind(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);
}
//消息发布
void onBasicPublish(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);
}
//消息确认
void onBasicAck(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);
}
//队列消息订阅
void onBasicConsume(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);
}
//队列消息取消订阅
void onBasicCancel(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);
}

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(_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

1. 封装连接管理类

2. 封装对外管理的类

3. 服务器管理

4. 服务器完整代码

相关推荐

评论抢沙发

评论前必须登录！

热门标签

置顶推荐

热门文章

最新文章

1. 封装连接管理类

2. 封装对外管理的类

3. 服务器管理

4. 服务器完整代码

相关推荐

评论 抢沙发

评论前必须登录！

热门标签

置顶推荐

热门文章

最新文章

评论抢沙发