一、为何要有这个模块？

在我们项目的前几弹当中，我们已经实现了：
交换机数据管理模块，队列数据管理模块，绑定信息管理模块，消息管理模块，虚拟机管理模块和路由匹配模块

二、消费者是否需要持久化？

在第六弹介绍虚拟机设计的时候，我们提到过，我们的项目模块划分：

消费者管理模块是描述模块，是对用户的描述，服务重启时，所有的TCP连接都要断开（我们暂且不考虑TCP的允许断线重连的机制），也就是说一旦服务重启，所有的消费者都会没有了

因此即使把消费者持久化，并成功恢复了，对应的用户也早就不存在了，因此持久化没有意义

三、怎么设计？

怎么设计？跟交换机那里一样，先抽象描述
在放到具体数据结构当中进行组织，当然，都要配上对应的访问操作接口

1.如何抽象描述？

1.回想一下基于生产消费模型的线程池

class threadpool
{
public:using functor = std::function<void()>;threadpool(int thread_num = 5){thread_vec.resize(thread_num);//消费者需要有一个线程函数[routine : 例行程序]for (int i = 0; i < thread_num; i++)thread_vec[i] = thread(std::bind(&threadpool::take, this));}// 生产者放数据void put(functor func);private:// 消费者拿数据void take();queue<functor> func_pool;  // 任务队列vector<thread> thread_vec; // 线程vec
};

在这个线程池当中,我们可以将消费者(工作线程)抽象为:

1. 线程标识(在这里是 : 下标)
2. 例行程序(在这里是 : take)

但是他并不符合我们的预期，因为：

1. 我们的消息队列是基于消息订阅的，消费者需要选择他想订阅的队列
2. 我们的消费者是可以用自己的想法来处理消息的
   【这才是消息中间件的灵魂】
3. 我们的消息是基于发布确认机制的，消费者成功消费消息之后要对消息进行确认【ack】，此时我们才会删除那个消息

2.如何组织

那么该如何组织消费者呢？
首先，消费者是依附于队列而存在的，通过订阅与取消订阅跟对应的队列建立联系，因此势必需要队列名这一字段作为成员

其次，因为不同虚拟机当中可以存在同名队列，而且我们的消费者管理模块并不属于虚拟机管理模块当中，因此势必需要虚拟机名称这一字段作为成员来区分同名队列

因为虚拟机管理模块是资源模块，而消费者管理模块是描述模块，两者不适合整合为统一的一个模块，持久化要求也不同

所以两者通过虚拟机名称这一字段进行关联。从而实现高内聚，低耦合

这也是借助MySQL数据库表的连接相关设计知识来设计出的模块组织方案

3.消息处理与确认问题的解决

我们的消费者是可以用自己的想法来处理消息的

因此我们可以让消费者自己提供消息处理函数，所以消费者类当中要有一个回调函数

message BasicProperities
{string msg_id = 1;DeliveryMode mode = 2;string routing_key = 3;
}
message Message 
{message ValidLoad{string body = 1;BasicProperities properities = 2;string valid = 3;}ValidLoad valid = 1;uint64 offset = 2;uint64 len = 3;  
}

那这个回调函数的签名是什么样的？
我们就要从消费者需要知道消息的什么信息开始：

1. 怎么处理这个消息
   需要知道消息的内容是什么，是谁处理的：

const std::string &body
const std::string &consumer_tag是consumer_tag处理了body这个消息

2. 怎么确认这个消息

bool basicAck(const std::string &qname, const std::string &msg_id);

msg_id在BasicProperities当中，因此我们需要：

const BasicProperites* bp

那么这个队列名呢？
在回调函数当中其实是不需要的，至于为何，我们以后会介绍的

因此：
回调函数的签名：

using ConsumerCallback = std::function<void(const std::string &tag, const ns_proto::BasicProperites *bp, const std::string &body)>;

4.自动确认标志

有些消息不是非常重要，且直到我们消费者能够进行确认时，中间可能相隔时间较长【毕竟肯定需要走一个网络IO】

所以RabbitMQ提出一个自动确认标志，来让发布确认机制更加灵活
bool auto_ack;

5.消费者代码

/*
因为我们的消费者是一种描述模块，而不是资源模块，其存在是为了更好的描述和管理消费者
*/
using ConsumerCallback = std::function<void(const std::string &, const ns_proto::BasicProperities *, const std::string &)>;struct Consumer
{using ptr = std::shared_ptr<Consumer>;Consumer() = default;Consumer(const std::string &tag, const ConsumerCallback &callback, const std::string &vhost_name, const std::string &qname, bool auto_ack): _consumer_tag(tag), _callback(callback), _vhost_name(vhost_name), _qname(qname), _auto_ack(auto_ack) {}std::string _consumer_tag;  // 消费者tag(唯一标识)ConsumerCallback _callback; // 消费者回调函数std::string _vhost_name;    // 消费者队列所在虚拟机名称std::string _qname;         // 消费者订阅的队列bool _auto_ack;             // 自动确认标志
};

2.队列消费者管理模块

1.为何要有队列消费者管理模块?

因为我们的消费者描述是依附于队列而存在的，只有订阅了某个队列的客户端才是消费者

类似于我们之前说的负载均衡，只不过消费者这里的负载均衡是这样设计并实现的：

2.队列消费者管理模块的设计

我们队列消费者管理模块要实现负载均衡：

1. RR轮转：将请求按顺序分配给各个消费者【实现简单，适用于消费者消费能力大致相同的场景】
2. 最小连接数：将请求分配给负载最低的消费者【完全式的负载均衡】

还有很多其他算法，大家感兴趣的话可以自行去了解一下，这里就不赘述了

为了降低消费者管理模块和消费者之间的耦合度，我们采用RR轮转，而不是最少连接数

这样的话，我们的消费者管理模块只负责组织维护对应队列的消费者，负载均衡式选取适合推送消息的消费者
之后消费者如何处理，处理多长时间，当前的状态是什么，我们无需考虑

【因为要支持轮询序号的随机访问，所以用vector
尽管vector在中间位置/起始位置删除的代价较大，但是消费者的获取需求远高于增删
所以在权衡下，选择了vector以牺牲新增和删除效率来提升查询效率】

成员：

0. 虚拟机名称
1. 队列名
2. vector<消费者句柄> consumer_vec
3. size_t seq；//轮询序号
4. 互斥锁【因为会存在多线程同时访问consumer_vec和int seq】

其实int seq可以通过搞成atomic<int>来确保线程安全，因为只需要++即可

size_t index=seq.fetch_add(1);
index%=consumer_vec.size();
反正index溢出不报错，直接回归到0也没什么大问题

但是因为都已经需要加锁保护consumer_vec了，所以不用原子类，保护这个临界资源也只是顺手的事
用了原子类，反而多此一举，纯属浪费资源

接口：

1. 新增消费者
2. 删除消费者
3. 获取消费者【RR轮转，负载均衡式获取】
4. 判断指定消费者是否存在
5. 销毁该队列所有消费者
6. 判断是否有消费者

3.总体消费者管理模块

其实就是增，删，查

成员：

1. unordered_map<pair<虚拟机名称，队列名>，队列消费者管理模块句柄>
2. 互斥锁

接口：
3. 初始化队列消费者管理模块
4. 销毁队列消费者管理模块
5. 新增消费者
6. 删除消费者
7. 获取消费者（RR轮转式负载均衡）
8. 判断指定消费者是否存在
9. 销毁所有消费者
10. 判断某一队列是否有消费者

四、代码

其实消费者管理模块就是设计起来略显复杂，但是代码简洁

1.队列消费者管理模块

总体上没啥难的,就是注意一下：
只有想要订阅当前队列消息的消费者才会调用该队列消费者管理模块的createConsumer函数
因此该函数无需传入vhost_name、qname作为参数

class QueueConsumerManager
{
public:using ptr = std::shared_ptr<QueueConsumerManager>;QueueConsumerManager(const std::string &vhost_name, const std::string &qname): _vhost_name(vhost_name), _qname(qname), _seq(0) {}// 1. 新增消费者[只有想要订阅当前队列消息的消费者才会调用该函数]Consumer::ptr createConsumer(const std::string &consumer_tag, const ConsumerCallback &callback, bool auto_ack){// 1. 加锁，并查找是否有该消费者std::unique_lock<std::mutex> ulock(_mutex);for (auto iter = _consumer_vec.begin(); iter != _consumer_vec.end(); ++iter){if ((*iter)->_consumer_tag == consumer_tag)return *iter;}// 2. 插入该消费者Consumer::ptr cp = std::make_shared<Consumer>(consumer_tag, callback, _vhost_name, _qname, auto_ack);_consumer_vec.push_back(cp);return cp;}void removeConsumer(const std::string &consumer_tag){// 加锁并删除该消费者std::unique_lock<std::mutex> ulock(_mutex);for (auto iter = _consumer_vec.begin(); iter != _consumer_vec.end(); ++iter){if ((*iter)->_consumer_tag == consumer_tag){_consumer_vec.erase(iter);return;}}}Consumer::ptr selectConsumer(){// 0. 加锁std::unique_lock<std::mutex> ulock(_mutex);if (_consumer_vec.empty()){default_warning("负载均衡式获取消费者出现问题，因为该队列的消费者为空");return Consumer::ptr();}int index = _seq++;_seq %= _consumer_vec.size();return _consumer_vec[index];}bool exist(const std::string &consumer_tag){std::unique_lock<std::mutex> ulock(_mutex);for (auto iter = _consumer_vec.begin(); iter != _consumer_vec.end(); ++iter){if ((*iter)->_consumer_tag == consumer_tag)return true;}return false;}bool empty(){std::unique_lock<std::mutex> ulock(_mutex);return _consumer_vec.empty();}void clear(){std::unique_lock<std::mutex> ulock(_mutex);_consumer_vec.clear();_seq = 0;}private:std::string _vhost_name;std::string _qname;std::mutex _mutex;std::vector<Consumer::ptr> _consumer_vec;size_t _seq;
};

2.总体消费者管理模块

1.pair<string,string>的哈希函数书写

我们采用BKDR字符串哈希函数，从而给pair<string,string>写出这样的哈希函数

struct HashFunc
{size_t hash(const std::string &str) const{size_t hash_val = 0;for (auto &e : str){hash_val *= 31; // BKDR哈希函数hash_val += e;}return hash_val;}size_t operator()(const std::pair<std::string, std::string> &elem) const{size_t hash1 = hash(elem.first), hash2 = hash(elem.second);// 使用两个字符串的哈希值，通过位运算和加法来生成最终的哈希值// 注意：这里使用了不同的质数来乘加，以减少哈希碰撞的可能性// 使用位运算和加法结合的方式// 使用了31和131作为乘数，它们都是常见的质数，常用于哈希函数中return (hash1 * 31 + hash2) * 131;}
};

BKDR哈希函数的介绍：

2.代码

就是增、删、复用

为何消息管理模块那里的key无需pair呢？无需存虚拟机名称呢？
因为消息管理模块在虚拟机模块当中，被虚拟机模块所整合了

class ConsumerManager
{
public:using ptr = std::shared_ptr<ConsumerManager>;void initQueueConsumerManager(const std::string &vhost_name, const std::string &qname){// 加锁,查找std::unique_lock<std::mutex> ulock(_mutex);std::pair<std::string, std::string> vqpair = {vhost_name, qname};if (_consumer_map.count(vqpair))return;_consumer_map.insert(std::make_pair(vqpair, std::make_shared<QueueConsumerManager>(vhost_name, qname)));}void destroyQueueConsumerManager(const std::string &vhost_name, const std::string &qname){std::unique_lock<std::mutex> ulock(_mutex);_consumer_map.erase({vhost_name, qname});}Consumer::ptr createConsumer(const std::string &vhost_name, const std::string &qname, const std::string &tag, const ConsumerCallback &callback, bool auto_ack){std::ostringstream oss;oss << "创建消费者失败，因为未能找到该队列消费者管理模块，qname = " << qname << "\n";QueueConsumerManager::ptr qcmp = getQueueConsumerManager(vhost_name, qname, oss);if (qcmp.get() != nullptr){return qcmp->createConsumer(tag, callback, auto_ack);}return Consumer::ptr();}bool removeConsumer(const std::string &vhost_name, const std::string &qname, const std::string &tag){std::ostringstream oss;oss << "删除消费者失败，因为未能找到该队列消费者管理模块，qname = " << qname << "\n";QueueConsumerManager::ptr qcmp = getQueueConsumerManager(vhost_name, qname, oss);if (qcmp.get() != nullptr){qcmp->removeConsumer(tag);return true;}return false;}Consumer::ptr selectConsumer(const std::string &vhost_name, const std::string &qname){std::ostringstream oss;oss << "负载均衡式获取消费者失败，因为未能找到该队列消费者管理模块，qname = " << qname << "\n";QueueConsumerManager::ptr qcmp = getQueueConsumerManager(vhost_name, qname, oss);if (qcmp.get() != nullptr){return qcmp->selectConsumer();}return Consumer::ptr();}bool exists(const std::string &vhost_name, const std::string &qname, const std::string &tag){std::ostringstream oss;oss << "判断是否存在消费者失败，因为未能找到该队列消费者管理模块，qname = " << qname << "\n";QueueConsumerManager::ptr qcmp = getQueueConsumerManager(vhost_name, qname, oss);if (qcmp.get() != nullptr){return qcmp->exist(tag);}return false;}bool empty(const std::string &vhost_name, const std::string &qname){std::ostringstream oss;oss << "判断队列消费者是否为空失败，因为未能找到该队列消费者管理模块，qname = " << qname << "\n";QueueConsumerManager::ptr qcmp = getQueueConsumerManager(vhost_name, qname, oss);if (qcmp.get() != nullptr){return qcmp->empty();}return false;}void clear(){std::unique_lock<std::mutex> ulock(_mutex);_consumer_map.clear();}QueueConsumerManager::ptr getQueueConsumerManager(const std::string &vhost_name, const std::string &qname, const std::ostringstream &oss){std::unique_lock<std::mutex> ulock(_mutex);auto iter = _consumer_map.find({vhost_name, qname});if (iter == _consumer_map.end()){default_error("%s",oss.str().c_str());return QueueConsumerManager::ptr();}return iter->second;}private:std::mutex _mutex;std::unordered_map<std::pair<std::string, std::string>, QueueConsumerManager::ptr, HashFunc> _consumer_map;
};

3.测试代码

#include "../mqserver/consumer.hpp"
#include <gtest/gtest.h>using namespace ns_mq;ConsumerManager::ptr cmp;class ConsumerTest : public testing::Environment
{
public:virtual void SetUp(){cmp = std::make_shared<ConsumerManager>();}virtual void TearDown(){cmp->clear();}
};TEST(consumer_test, init_test)
{cmp->initQueueConsumerManager("vhost1","queue1");cmp->initQueueConsumerManager("vhost1","queue2");cmp->initQueueConsumerManager("vhost1","queue3");std::ostringstream oss;ASSERT_NE(cmp->getQueueConsumerManager("vhost1","queue1", oss).get(), nullptr);ASSERT_NE(cmp->getQueueConsumerManager("vhost1","queue2", oss).get(), nullptr);ASSERT_NE(cmp->getQueueConsumerManager("vhost1","queue3", oss).get(), nullptr);
}TEST(consumer_test, insert_test)
{cmp->createConsumer("vhost1","queue1", "consumer1", nullptr, false);cmp->createConsumer("vhost1","queue1", "consumer2", nullptr, false);cmp->createConsumer("vhost1","queue1", "consumer3", nullptr, false);cmp->createConsumer("vhost1","queue1", "consumer4", nullptr, false);ASSERT_EQ(cmp->exists("vhost1","queue1", "consumer1"), true);ASSERT_EQ(cmp->exists("vhost1","queue1", "consumer2"), true);ASSERT_EQ(cmp->exists("vhost1","queue1", "consumer3"), true);ASSERT_EQ(cmp->exists("vhost1","queue1", "consumer4"), true);
}TEST(consumer_test, select_test)
{Consumer::ptr cp = cmp->selectConsumer("vhost1","queue1");ASSERT_EQ(cp->_consumer_tag, std::string("consumer1"));cp = cmp->selectConsumer("vhost1","queue1");ASSERT_EQ(cp->_consumer_tag, std::string("consumer2"));cp = cmp->selectConsumer("vhost1","queue1");ASSERT_EQ(cp->_consumer_tag, std::string("consumer3"));cp = cmp->selectConsumer("vhost1","queue1");ASSERT_EQ(cp->_consumer_tag, std::string("consumer4"));
}TEST(consumer_test, erase_test)
{cmp->removeConsumer("vhost1","queue1", "consumer1");cmp->removeConsumer("vhost1","queue1", "consumer2");ASSERT_EQ(cmp->exists("vhost1","queue1", "consumer1"), false);ASSERT_EQ(cmp->exists("vhost1","queue1", "consumer2"), false);ASSERT_EQ(cmp->exists("vhost1","queue1", "consumer3"), true);ASSERT_EQ(cmp->exists("vhost1","queue1", "consumer4"), true);
}TEST(consumer_test, destroy_test)
{cmp->destroyQueueConsumerManager("vhost1","queue1");cmp->destroyQueueConsumerManager("vhost1","queue2");cmp->destroyQueueConsumerManager("vhost1","queue3");std::ostringstream oss;ASSERT_EQ(cmp->getQueueConsumerManager("vhost1","queue1", oss).get(), nullptr);ASSERT_EQ(cmp->getQueueConsumerManager("vhost1","queue2", oss).get(), nullptr);ASSERT_EQ(cmp->getQueueConsumerManager("vhost1","queue3", oss).get(), nullptr);
}int main(int argc, char *argv[])
{testing::AddGlobalTestEnvironment(new ConsumerTest);testing::InitGoogleTest(&argc, argv);return RUN_ALL_TESTS();
}

以上就是项目第七弹：消费者管理模块的全部内容