⽹络通信协议设计

其中⽣产者和消费者都是客⼾端，它们都需要通过⽹络和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;
}