提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档
文章目录
- 前言
- 消费者字段
- 消费者内存管理类
- 新增/删除消费者
- 总的消费者管理类
前言
只有订阅客户端订阅了一个队列消息后，他才是一个消费者，而消费者存在的意义是，当这个队列上有消息的时候我们呢需要给订阅了这个队列的客户端推送消息，也就需要获取客户端的连接，所以我们呢可以通过消费者来找到对应的连接来进行消息的推送。
消费者字段
消费者订阅的队列名称，消费者也是按照队列问单元进行管理的。
消费者标志，标志消费者唯一性
自动应答标志，客户端在订阅队列消息时可以进行设置，如果为true，则将消息推送给消费客户端后不需要等待应带，直接将消息进行删除。
还有一个回调函数。当队列有消息到来时，就需要选择一个队列消费者进行消费，那么如何进行消费呢？就是调用这个回调函数。这个函数的逻辑固定，就是通过连接发送消息。是由我们broker服务器设置的。
//当队列收到一条消息后，需要选择一个消费者进行消费，那么如何消费呢？//就是调用这个回调函数，其内部逻辑是：找到消费者对应的连接，将消息发送给客户端。using ConsumerCallBack = std::function<void(const std::string&,const BasicProperties *,const std::string &)>;struct Consumer{using ptr = std::shared_ptr<Consumer>;std::string _qname; //消费者订阅的队列名称std::string _ctag; //消费者标识bool _auto_ack; //自动应答标志ConsumerCallBack _cb;
}
消费者内存管理类
消费者是以队列为单位进行管理的，我们使用一个vector管理订阅该队列的消费者。
目的是方便进行rr轮转。
class QueueConsumer{private:std::string _qname; //队列名称std::mutex _mutex; uint64_t _rr_sep; //轮转序号std::vector<Consumer::ptr> _consumers;}
新增/删除消费者
构造一个消费者对象进行添加。删除就是遍历删除。
//新增一个消费者 Consumer::ptr create(const std::string &qname,const std::string &ctag,bool auto_ack,const ConsumerCallBack &cb){//1.加锁std::unique_lock<std::mutex> lock(_mutex);//2.判断消费者是否存在for(auto &cons : _consumers){if(cons->_ctag == ctag){return cons;}}//3.创建消费者并添加管理Consumer::ptr csp = std::make_shared<Consumer>(qname,ctag,auto_ack,cb);_consumers.push_back(csp);return csp;
}
//删除一个消费者：取消订阅、信道关闭、连接关闭的时候删除
void remove(const std::string &ctag){//1.加锁std::unique_lock<std::mutex> lock(_mutex);//2.遍历删除for(auto it = _consumers.begin(); it != _consumers.end(); ++it){if((*it)->_ctag == ctag){_consumers.erase(it);return;}}return;
}
选择一个消费者，rr轮转的思想。
在收到一条消息后，会进行路由匹配，获取到匹配成功的队列。然后把消息推送到队列上，同时取出一个消费者进行消费。
//选择一个消费者，rr轮转的思想Consumer::ptr choose(){//1.加锁std::unique_lock<std::mutex> lock(_mutex);//如果没有消费者，则返回空对象if(_consumers.size() == 0){return Consumer::ptr();}//2.获取当前选择的消费者下标uint64_t idx = _rr_sep % _consumers.size();_rr_sep++;//3.获取对象并返回return _consumers[idx];}
总的消费者管理类
这个类才是对外提供的类。
就是用一个哈希表把队列名称和消费者管理对象关联起来进行管理。
有一initQueueConsumer函数就是用力啊初始化队列消费者管理对象的，需要判断当前以存在的队列，来进行初始化对应的消费者管理类。
class ConsumerManager{private: std::mutex _mutex;std::unordered_map<std::string,QueueConsumer::ptr> _qconsumers;public:using ptr = std::shared_ptr<ConsumerManager>;ConsumerManager(){}//初始化队列消费者管理void initQueueConsumer(const std::string &qname){//1. 加锁std::unique_lock<std::mutex> lock(_mutex);//2. 重复判断auto it = _qconsumers.find(qname);if (it != _qconsumers.end()) {return ;}//3. 新增auto qconsumers = std::make_shared<QueueConsumer>(qname);_qconsumers.insert(std::make_pair(qname, qconsumers));}
}
删除队列消费者管理类对象，当一个队列删除时，就选哦删除这个队列的消费者管理类对象，
//删除队列消费者管理void destroyQueueConsumer(const std::string &qname){std::unique_lock<std::mutex> lock(_mutex);auto it = _qconsumers.find(qname);if (it == _qconsumers.end()) {return ;}_qconsumers.erase(it);}
创建/删除指定队列的消费者管理类对象中的消费者
根据队列的名称，选择指定的消费者管理类对象进行一个操作即可，
Consumer::ptr create(const std::string &qname,const std::string &ctag,bool auto_ack,const ConsumerCallBack &cb){QueueConsumer::ptr qcp;{std::unique_lock<std::mutex> lock(_mutex);auto it = _qconsumers.find(qname);if(it == _qconsumers.end()){ELOG("没有找到队列 %s 的消费者管理句柄！",qname.c_str());return Consumer::ptr();}qcp = it->second;}//这个create不需要我们这里的mutex保护，他有自己的mutexreturn qcp->create(qname,ctag,auto_ack,cb);
}
void remove(const std::string &ctag,const std::string &qname){QueueConsumer::ptr qcp;{std::unique_lock<std::mutex> lock(_mutex);auto it = _qconsumers.find(qname);if(it == _qconsumers.end()){ELOG("没有找到队列 %s 的消费者管理句柄！",qname.c_str());return ;}qcp = it->second;}qcp->remove(ctag);
}
选择指定队列的消费者管理类对象中的消费者
Consumer::ptr choose(const std::string &qname){QueueConsumer::ptr qcp;{std::unique_lock<std::mutex> lock(_mutex);auto it = _qconsumers.find(qname);if(it == _qconsumers.end()){ELOG("没有找到队列 %s 的消费者管理句柄！",qname.c_str());return Consumer::ptr();}qcp = it->second;}return qcp->choose();
}