滕州微信网站,获取别人wordpress主题,沈阳男科医院排名前十,企业网站规范提示#xff1a;文章写完后#xff0c;目录可以自动生成#xff0c;如何生成可参考右边的帮助文档 文章目录 前言消费者模块信道管理模块管理的字段提供的接口 信道内存管理连接管理类 前言
在RabbitMQ中#xff0c;提供服务的是信道#xff0c;因此在客⼾端的实现中文章写完后目录可以自动生成如何生成可参考右边的帮助文档 文章目录 前言消费者模块信道管理模块管理的字段提供的接口 信道内存管理连接管理类 前言
在RabbitMQ中提供服务的是信道因此在客⼾端的实现中弱化了Client客⼾端的概念。在RabbitMQ中并不会向⽤⼾展⽰⽹络通信的概念出来⽽是以⼀种提供服务的形式来体现。实现思想类似于普通的功能接⼝封装⼀个接⼝实现⼀个功能接⼝内部完成向客⼾端请求的过程但是对外并不需要体现出客⼾端与服务端通信的概念⽤⼾需要什么服务就调⽤什么接⼝就⾏。 消费者模块
客户端这块对于消费者模块是不需要管理的当进行消息订阅的时候就会创建出一个消费者而这消费者的作用就是: • 描述当前信道订阅了哪个队列的消息。 • 描述了收到消息后该如何对这条消息进⾏处理。 • 描述收到消息后是否需要进⾏确认回复。
using ConsumerCallBack std::functionvoid(const std::string,BasicProperties *,const std::string );struct Consumer{using ptr std::shared_ptrConsumer;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_mapstd::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_lockstd::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_mapstd::string,Channel::ptr _channels;}连接管理类
在客⼾端这边RabbitMQ弱化了客⼾端的概念因为⽤⼾所需的服务都是通过信道来提供的因此操作思想转换为先创建连接通过连接创建信道通过信道提供服务这⼀流程。 这个模块同样是针对muduo库客⼾端连接的⼆次封装向⽤⼾提供创建channel信道的接⼝创建信道后可以通过信道来获取指定服务。
class Connection {public:using ptr std::shared_ptrConnection;
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());
}
