培训机构整顿_安卓优化大师最新版下载_网站自动提交收录_免费发广告的软件

⽹络通信协议设计

其中⽣产者和消费者都是客⼾端，它们都需要通过⽹络和BrokerServer进⾏通信。具体通信的过程我们使⽤Muduo库来实现，使⽤TCP作为通信的底层协议，同时在这个基础上⾃定义应⽤层协议，完成客⼾端对服务器功能的远端调⽤。我们要实现的远端调⽤接⼝包括：

使⽤⼆进制的⽅式设计应⽤层协议。因为MQMessage的消息体是使⽤Protobuf进⾏序列化的，本⾝是按照⼆进制存储的，所以不太适合⽤json等⽂本格式来定义协议。

下⾯我们设计⼀下应⽤层协议：请求/响应报⽂设计

• len：4个字节,表⽰整个报⽂的⻓度
• nameLen:4个字节，表⽰typeName数组的⻓度
• typeName：是个字节数组，占nameLen个字节,表⽰请求/响应报⽂的类型名，作⽤是分发不同消息到对应的远端接⼝调⽤中
• protobufData：是个字节数组，占len-nameLen-8个字节，表⽰请求/响应参数数据通过protobuf序列化之后的⼆进制
• checkSum：4个字节，表⽰整个消息的校验和，作⽤是为了校验请求/响应报⽂的完整性

⼀个创建交换机的请求，如下图⽰：

信道管理模块

在AMQP模型中，除了通信连接Connection概念外，还有⼀个Channel的概念，Channel是针对Connection连接的⼀个更细粒度的通信信道，多个Channel可以使⽤同⼀个通信连接Connection进⾏通信，但是同⼀个Connection的Channel之间相互独⽴。

⽽信道模块就是再次将上述模块进⾏整合提供服务的模块

proto.proto

syntax = "proto3";
package nzq;import "msg.proto";//信道的打开与关闭
message openChannelRequest{string rid = 1;string cid = 2;
};
message closeChannelRequest{string rid = 1;string cid = 2;
};
//交换机的声明与删除
message declareExchangeRequest{string rid = 1;string cid = 2;string exchange_name = 3;ExchangeType exchange_type = 4;bool durable = 5;bool auto_delete = 6;map<string, string> args = 7;
};
message deleteExchangeRequest{string rid = 1;string cid = 2;string exchange_name = 3;
};
//队列的声明与删除
message declareQueueRequest{string rid = 1;string cid = 2;string queue_name = 3;bool exclusive = 4;bool durable = 5;bool auto_delete = 6;map<string, string> args = 7;
};
message deleteQueueRequest{string rid = 1;string cid = 2;string queue_name = 3;
};
//队列的绑定与解除绑定
message queueBindRequest{string rid = 1;string cid = 2;string exchange_name = 3;string queue_name = 4;string binding_key = 5;
};
message queueUnBindRequest{string rid = 1;string cid = 2;string exchange_name = 3;string queue_name = 4;
};
//消息的发布
message basicPublishRequest {string rid = 1;string cid = 2;string exchange_name = 3;string body = 4;BasicProperties properties = 5;
};
//消息的确认
message basicAckRequest {string rid = 1;string cid = 2;string queue_name = 3;string message_id = 4;
};
//队列的订阅
message basicConsumeRequest {string rid = 1;string cid = 2;string consumer_tag  =3;string queue_name = 4;bool auto_ack = 5;
};
//订阅的取消
message basicCancelRequest {string rid = 1;string cid = 2;string consumer_tag = 3;string queue_name = 4;
};
//消息的推送
message basicConsumeResponse {string cid = 1;string consumer_tag = 2;string body = 3;BasicProperties properties = 4;
};
//通用响应
message basicCommonResponse {string rid = 1;string cid = 2;bool ok = 3;
}

1. 管理信息：

• a. 信道ID：信道的唯⼀标识
• b. 信道关联的消费者：⽤于消费者信道在关闭的时候取消订阅，删除订阅者信息
• c. 信道关联的连接：⽤于向客⼾端发送数据（响应，推送的消息）
• d. protobuf协议处理句柄：⽹络通信前的协议处理
• e. 消费者管理句柄：信道关闭/取消订阅的时候，通过句柄删除订阅者信息
• f. 虚拟机句柄：交换机/队列/绑定/消息数据管理
• g. ⼯作线程池句柄（⼀条消息被发布到队列后，需要将消息推送给订阅了对应队列的消费者，过
  程由线程池完成）

2. 管理操作：

• a. 提供声明&删除交换机操作（删除交换机的同时删除交换机关联的绑定信息）
• b. 提供声明&删除队列操作（删除队列的同时，删除队列关联的绑定信息，消息，消费者信息）
• c. 提供绑定&解绑队列操作
• d. 提供订阅&取消订阅队列消息操作
• e. 提供发布&确认消息操作

    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;using openChannelRequestPtr = std::shared_ptr<openChannelRequest>;using closeChannelRequestPtr = std::shared_ptr<closeChannelRequest>;using declareExchangeRequestPtr = std::shared_ptr<declareExchangeRequest>;using deleteExchangeRequestPtr = std::shared_ptr<deleteExchangeRequest>;using declareQueueRequestPtr = std::shared_ptr<declareQueueRequest>;using deleteQueueRequestPtr = std::shared_ptr<deleteQueueRequest>;using queueBindRequestPtr = std::shared_ptr<queueBindRequest>;using queueUnBindRequestPtr = std::shared_ptr<queueUnBindRequest>;using basicPublishRequestPtr = std::shared_ptr<basicPublishRequest>;using basicAckRequestPtr = std::shared_ptr<basicAckRequest>;using basicConsumeRequestPtr = std::shared_ptr<basicConsumeRequest>;using basicCancelRequestPtr = std::shared_ptr<basicCancelRequest>;class Channel {public:using ptr = std::shared_ptr<Channel>;Channel(const std::string &id, const VirtualHost::ptr &host, const ConsumerManager::ptr &cmp, const ProtobufCodecPtr &codec, const muduo::net::TcpConnectionPtr &conn,const threadpool::ptr &pool):_cid(id),_conn(conn),_codec(codec),_cmp(cmp),_host(host),_pool(pool){DLOG("new Channel: %p", this);}~Channel() {if (_consumer.get() != nullptr) {_cmp->remove(_consumer->tag, _consumer->qname);}DLOG("del Channel: %p", this);}//交换机的声明与删除void declareExchange(const declareExchangeRequestPtr &req) {bool ret = _host->declareExchange(req->exchange_name(), req->exchange_type(), req->durable(), req->auto_delete(), req->args());return basicResponse(ret, req->rid(), req->cid());}void deleteExchange(const deleteExchangeRequestPtr &req) {_host->deleteExchange(req->exchange_name());return basicResponse(true, req->rid(), req->cid());}//队列的声明与删除void declareQueue(const declareQueueRequestPtr &req) {bool ret = _host->declareQueue(req->queue_name(),req->durable(), req->exclusive(),req->auto_delete(), req->args());if (ret == false) {return basicResponse(false, req->rid(), req->cid());}_cmp->initQueueConsumer(req->queue_name());//初始化队列的消费者管理句柄return basicResponse(true, req->rid(), req->cid());}void deleteQueue(const deleteQueueRequestPtr &req) {_cmp->destroyQueueConsumer(req->queue_name());_host->deleteQueue(req->queue_name());return basicResponse(true, req->rid(), req->cid());}//队列的绑定与解除绑定void queueBind(const queueBindRequestPtr &req) {bool ret = _host->bind(req->exchange_name(), req->queue_name(), req->binding_key());return basicResponse(ret, req->rid(), req->cid());}void queueUnBind(const queueUnBindRequestPtr &req) {_host->unBind(req->exchange_name(), req->queue_name());return basicResponse(true, req->rid(), req->cid());}//消息的发布void basicPublish(const basicPublishRequestPtr &req) {//1. 判断交换机是否存在auto ep = _host->selectExchange(req->exchange_name());if (ep.get() == nullptr) {return basicResponse(false, req->rid(), req->cid());}//2. 进行交换路由，判断消息可以发布到交换机绑定的哪个队列中MsgQueueBindingMap mqbm = _host->exchangeBindings(req->exchange_name());BasicProperties *properties = nullptr;std::string routing_key;if (req->has_properties()) {properties = req->mutable_properties();routing_key = properties->routing_key();}for (auto &binding : mqbm) {if (Router::route(ep->type, routing_key, binding.second->binding_key)) {//3. 将消息添加到队列中（添加消息的管理）_host->basicPublish(binding.first, properties, req->body());//4. 向线程池中添加一个消息消费任务（向指定队列的订阅者去推送消息--线程池完成）auto task = std::bind(&Channel::consume, this, binding.first);_pool->push(task);}}return basicResponse(true, req->rid(), req->cid());}//消息的确认void basicAck(const basicAckRequestPtr &req) {_host->basicAck(req->queue_name(), req->message_id());return basicResponse(true, req->rid(), req->cid());}//订阅队列消息void basicConsume(const basicConsumeRequestPtr &req) {//1. 判断队列是否存在bool ret = _host->existsQueue(req->queue_name());if (ret == false) {return basicResponse(false, req->rid(), req->cid());}//2. 创建队列的消费者auto cb = std::bind(&Channel::callback, this, std::placeholders::_1,std::placeholders::_2, std::placeholders::_3);//创建了消费者之后，当前的channel角色就是个消费者_consumer = _cmp->create(req->consumer_tag(), req->queue_name(), req->auto_ack(), cb);return basicResponse(true, req->rid(), req->cid());}//取消订阅void basicCancel(const basicCancelRequestPtr &req) {_cmp->remove(req->consumer_tag(), req->queue_name());return basicResponse(true, req->rid(), req->cid());}private:void callback(const std::string tag, const BasicProperties *bp, const std::string &body) {//针对参数组织出推送消息请求，将消息推送给channel对应的客户端basicConsumeResponse resp;resp.set_cid(_cid);resp.set_body(body);resp.set_consumer_tag(tag);if (bp) {resp.mutable_properties()->set_id(bp->id());resp.mutable_properties()->set_delivery_mode(bp->delivery_mode());resp.mutable_properties()->set_routing_key(bp->routing_key());}_codec->send(_conn, resp);}void consume(const std::string &qname) {//指定队列消费消息//1. 从队列中取出一条消息MessagePtr mp = _host->basicConsume(qname);if (mp.get() == nullptr) {DLOG("执行消费任务失败，%s 队列没有消息！", qname.c_str());return;}//2. 从队列订阅者中取出一个订阅者Consumer::ptr cp = _cmp->choose(qname);if (cp.get() == nullptr) {DLOG("执行消费任务失败，%s 队列没有消费者！", qname.c_str());return;}//3. 调用订阅者对应的消息处理函数，实现消息的推送cp->callback(cp->tag, mp->mutable_payload()->mutable_properties(), mp->payload().body());//4. 判断如果订阅者是自动确认---不需要等待确认，直接删除消息，否则需要外部收到消息确认后再删除if (cp->auto_ack) _host->basicAck(qname, mp->payload().properties().id());}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:std::string _cid;Consumer::ptr _consumer;muduo::net::TcpConnectionPtr _conn;ProtobufCodecPtr _codec;ConsumerManager::ptr _cmp;VirtualHost::ptr _host;threadpool::ptr _pool;};

3. 信道管理

• a. 信道的增删查。

    class ChannelManager {public:using ptr = std::shared_ptr<ChannelManager>;ChannelManager(){}bool openChannel(const std::string &id, 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 = _channels.find(id);if (it != _channels.end()) {DLOG("信道：%s 已经存在!", id.c_str());return false;}auto channel = std::make_shared<Channel>(id, host, cmp, codec, conn, pool);_channels.insert(std::make_pair(id, channel));return true;}void closeChannel(const std::string &id){std::unique_lock<std::mutex> lock(_mutex);_channels.erase(id);}Channel::ptr getChannel(const std::string &id) {std::unique_lock<std::mutex> lock(_mutex);auto it = _channels.find(id);if (it == _channels.end()) {return Channel::ptr();}return it->second;}private:std::mutex _mutex;std::unordered_map<std::string, Channel::ptr> _channels;};

连接管理模块

向⽤⼾提供⼀个⽤于实现⽹络通信的Connection对象，从其内部可创建出粒度更轻的Channel对象，⽤于与客⼾端进⾏⽹络通信。

1. 成员信息：

• a. 连接关联的信道管理句柄（实现信道的增删查）
• b. 连接关联的实际⽤于通信的muduo::net::Connection连接
• c. protobuf协议处理的句柄（ProtobufCodec对象）
• d. 消费者管理句柄
• e. 虚拟机句柄
• f. 异步⼯作线程池句柄

2. 连接操作：

• a. 提供创建Channel信道的操作
• b. 提供删除Channel信道的操作

    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->cid(), _host, _cmp, _codec, _conn, _pool);if (ret == false) {DLOG("创建信道的时候，信道ID重复了");return basicResponse(false, req->rid(), req->cid());}DLOG("%s 信道创建成功！", req->cid().c_str());//3. 给客户端进行回复return basicResponse(true, req->rid(), req->cid());}void closeChannel(const closeChannelRequestPtr &req) {_channels->closeChannel(req->cid());return 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;};

3. 连接管理：

• a. 连接的增删查

    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;};

注意:
在RabbitMQ中,虚拟主机是可以随意创建/删除的，但是咱们此处为了实现简单,并没有实现虚拟主机的管理，因此我们默认就只有⼀个虚拟主机的存在，但是在数据结构的设计上我们预留了对于多虚拟主机的管理，从⽽保证不同虚拟主机中的Exchange、Queue、Binding、Message等资源都是相互隔离的。

服务器模块实现

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

BrokerServer模块是对整体服务器所有模块的整合，接收客⼾端的请求，并提供服务。

基于前边实现的简单的翻译服务器代码，进⾏改造，只需要实现服务器内部提供服务的各个业务接即可。

在各个业务处理函数中，也⽐较简单，创建信道后，每次请求过来后，找到请求对应的信道句柄，通过句柄调⽤前边封装好的处理接⼝进⾏请求处理，最终返回处理结果。

    #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<nzq::openChannelRequest>(std::bind(&Server::onOpenChannel, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_dispatcher.registerMessageCallback<nzq::closeChannelRequest>(std::bind(&Server::onCloseChannel, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_dispatcher.registerMessageCallback<nzq::declareExchangeRequest>(std::bind(&Server::onDeclareExchange, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_dispatcher.registerMessageCallback<nzq::deleteExchangeRequest>(std::bind(&Server::onDeleteExchange, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_dispatcher.registerMessageCallback<nzq::declareQueueRequest>(std::bind(&Server::onDeclareQueue, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_dispatcher.registerMessageCallback<nzq::deleteQueueRequest>(std::bind(&Server::onDeleteQueue, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_dispatcher.registerMessageCallback<nzq::queueBindRequest>(std::bind(&Server::onQueueBind, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_dispatcher.registerMessageCallback<nzq::queueUnBindRequest>(std::bind(&Server::onQueueUnBind, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_dispatcher.registerMessageCallback<nzq::basicPublishRequest>(std::bind(&Server::onBasicPublish, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_dispatcher.registerMessageCallback<nzq::basicAckRequest>(std::bind(&Server::onBasicAck, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_dispatcher.registerMessageCallback<nzq::basicConsumeRequest>(std::bind(&Server::onBasicConsume, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_dispatcher.registerMessageCallback<nzq::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;}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;}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;}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;}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;}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;}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;}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;}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;}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;}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;}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;}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;};

server.cc

#include "broker.hpp"int main()
{nzq::Server server(8085, "./data/");server.start();return 0;
}