rabbitMq------客户端模块

提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档

---

前言

在RabbitMQ中，提供服务的是信道，因此在客⼾端的实现中，弱化了Client客⼾端的概念。在RabbitMQ中并不会向⽤⼾展⽰⽹络通信的概念出来，⽽是以⼀种提供服务的形式来体现。实现思想类似于普通的功能接⼝封装，⼀个接⼝实现⼀个功能，接⼝内部完成向客⼾端请求的过程，但是对外并不需要体现出客⼾端与服务端通信的概念，⽤⼾需要什么服务就调⽤什么接⼝就⾏。

---

消费者模块

客户端这块对于消费者模块是不需要管理的，当进行消息订阅的时候，就会创建出一个消费者，而这消费者的作用就是:
• 描述当前信道订阅了哪个队列的消息。
• 描述了收到消息后该如何对这条消息进⾏处理。
• 描述收到消息后是否需要进⾏确认回复。

using ConsumerCallBack = std::function<void(const std::string&,BasicProperties *,const std::string &)>;struct Consumer{using ptr = std::shared_ptr<Consumer>;std::string _qname; //消费者订阅的队列名称std::string _ctag;  //消费者标识bool _auto_ack;  //自动应答标志ConsumerCallBack _cb;   }

信道管理模块

同样的客户端也有信道，其功能与服务端几乎一致，只不过客户端的信道是为用户提供服务的，而服务器的信道是为客户端的请求提供服务的。也可以理解是⽤⼾通过客⼾端channel的接⼝调⽤来向服务端发送对应请求，获取请求的服务。

管理的字段

客户端的信道需要管理的字段中有一个哈希表，是请求id和通用响应结构的映射。
因为在muduo库中发送和接收都是异步的，例如我们声明一个交换机，这个请求可能还在发送缓冲区，并没有发送，我们此时如果去给这个交换机推送消息就会出问题。因此需要我们⾃⼰在收到响应后，通过判断是否是等待的指定响应来进⾏同步。

class Channel{private:std::string _cid;muduo::net::TcpConnectionPtr _conn;ProtobufCodecPtr _codec; Consumer::ptr _consumer;//由于muduo库的发送和接收都是异步的，例如我们声明一个交换机，这个请求可能还在发送缓冲区，并没有发送，我们此时如果去给这个交换机推送消息就会出问题。因此需要我们⾃⼰在收到响应后，通过判断是否是等待的指定响应来进⾏同步std::mutex _mutex;std::condition_variable _cv;std::unordered_map<std::string,basicCommonResponsePtr> _basic_resp;}

提供的接口

客户端的channel和服务端的接口都是几乎一致的，客户端的接口中是组织请求，向服务端发起请求，服务端的接口是接收请求进行业务处理。

但客户端中的channel提供了打开信道和关闭信道这俩个接口，他们是向服务端发起请求，在服务器上创建信道。

//这两个接口是向服务端发送请求，在服务端创建对应信道bool openChannel(){//组织请求openChannelRequest req;std::string rid = UUIDHelper::uuid();req.set_rid(rid);req.set_cid(_cid);//发送_codec->send(_conn,req);basicCommonResponsePtr resp = waitResponse(rid);return resp->ok();}void closeChannel(){//组织请求closeChannelRequest req;std::string rid = UUIDHelper::uuid();req.set_rid(rid);req.set_cid(_cid);//发送_codec->send(_conn,req);waitResponse(rid);return;}

其中每个接口中都需要调用waitResponse，用来同步等待响应。

basicCommonResponsePtr waitResponse(std::string &rid){std::unique_lock<std::mutex> lock(_mutex);_cv.wait(lock,[&rid,this](){return _basic_resp.find(rid) != _basic_resp.end();});basicCommonResponsePtr basic_resp = _basic_resp[rid];_basic_resp.erase(rid);return basic_resp;}

信道内存管理

对信道的一个总的管理类。
有一个哈希表，是信道ID和信道对象的映射。
他提供了三个接口，创建信道/删除信道/获取指定信道。

class ChannelManager{private:std::mutex _mutex;std::unordered_map<std::string,Channel::ptr> _channels;}

连接管理类

在客⼾端这边，RabbitMQ弱化了客⼾端的概念，因为⽤⼾所需的服务都是通过信道来提供的，因此操作思想转换为先创建连接，通过连接创建信道，通过信道提供服务这⼀流程。
这个模块同样是针对muduo库客⼾端连接的⼆次封装，向⽤⼾提供创建channel信道的接⼝，创建信道后，可以通过信道来获取指定服务。

class Connection {public:using ptr = std::shared_ptr<Connection>;
private:muduo::CountDownLatch _latch;//实现同步的muduo::net::TcpConnectionPtr _conn;//客户端对应的连接muduo::net::TcpClient _client;//客户端ProtobufDispatcher _dispatcher;//请求分发器ProtobufCodecPtr _codec;//协议处理器AsyncWorker::ptr _worker;ChannelManager::ptr _channel_manager;}

在客户端这边，会收到两种响应，一种是基础响应，一种是消息推送响应。
我们需要注册对于这两种类型响应的回调函数。

基础响应，在基础响应中有一个cid字段，根据cid获取指定的信道对象，
调用信道对象中的putBasicResponse接口。
也就是往信道的hashMap中添加响应值。

void basicResponse(const muduo::net::TcpConnectionPtr& conn, const basicCommonResponsePtr& message, muduo::Timestamp) {//1. 找到信道Channel::ptr channel = _channel_manager->get(message->cid());if (channel.get() == nullptr) {DLOG("未找到信道信息！");return;}//2. 将得到的响应对象，添加到信道的基础响应hash_map中channel->putBasicResponse(message);
}

消息推送，在收到消息推送的响应后，需要更具响应中的rid，获取指定的信道对象，然后封装一个任务，这个任务就是调用信道中的consume接口，这个接口就是执行消费者中设置的回调函数。我们把这个任务放入到线程池中。

void consumeResponse(const muduo::net::TcpConnectionPtr& conn, const basicConsumerResponsePtr& message, muduo::Timestamp){
//1. 找到信道Channel::ptr channel = _channel_manager->get(message->cid());if (channel.get() == nullptr) {DLOG("未找到信道信息！");return;}//2. 封装异步任务（消息处理任务），抛入线程池_worker->pool.push([channel, message](){channel->consume(message);});
}

连接管理提供了两个接口，打开信道和关闭信道。这是给用户提供的，用来创建一个信道，通过信道进行服务的调用。

 Channel::ptr openChannel() {Channel::ptr channel = _channel_manager->create(_conn, _codec);bool ret = channel->openChannel();if (ret == false) {DLOG("打开信道失败！");return Channel::ptr();}return channel;
}void closeChannel(const Channel::ptr &channel) {channel->closeChannel();_channel_manager->remove(channel->cid());
}