一、为何要设计信道池

1.引入信道的好处

RabbitMQ为了实现多线程之间的资源可见性的约束和隔离，在网络服务连接和业务模块之间加入信道一层，由信道负责提供所有服务，线程想要申请服务必须在线程函数内创建信道

因此通过利用线程独占栈空间来限制句柄资源的可见性，从而在技术上实现了多线程间资源可见性的约束和隔离

用多线程术语来说的话：
将连接细分为信道就是将连接这种临界资源细分为了信道这种非临界资源，是一种资源划分/预订机制，跟信号量的原理一样

2.为何要设计信道池

信道的创建比起连接更为轻量，一定程度上复用了TCP连接
但是信道的频繁创建与销毁的开销依然还是存在的，特别是在高并发场景下，其消耗不容忽视

因此，借助池化技术的思想，我们可以实现一个信道池，提前创建一批信道，这样就能提高客户端创建信道的速度，并且进行了信道级别的复用

在高并发场景下，这种效率的提升是显而易见的

二、信道池的设计

我们之前说过，连接就像是学校食堂的某一个窗口，信道就是对应窗口的号码牌，我们排在对应窗口的学生就是线程

那如何用编程将号码牌设计为“号码牌”池呢？

1.服务器需要设计信道池吗？

服务器无需设计信道池，因为：

1. 客户端的信道跟服务器的信道是1比1的，只要客户端的信道是可复用的，那么服务器的信道就是可复用的
2. 一个服务器可以连接多个客户端，因此服务器的初始信道数不好给定

2.设计：动态变化的信道池

1.为什么？

因为流量存在高峰期和低谷期，而在高峰期时某个客户端可能会申请大量信道，过了高峰期之后，大量信道本就已经不再使用了，但是依然占用资源

在这段时间内，信道资源一直占用内存，严重情况下会拖慢客户端，服务器的效率，因此我们的信道池是动态变化的

这样才能更好的适应流量的高峰和低谷期，体现了RabbitMQ的削峰填谷

【动态信道池能够根据当前的流量需求自动调整信道的数量和状态。
在高峰期，系统可以增加信道数量以满足高并发需求；
而在低谷期，则通过回收不再使用的信道来释放资源，避免不必要的内存占用】

2.怎么办？

1.动态扩容和缩容

扩容：我们就一次扩2倍吧，基础值是4

缩容：
什么时候需要缩容？
信道池当中有些信道长时间不被使用时，我们就将对应信道给关闭并删除掉

因此我们需要一个定时器，实现【超时缩容机制】，muduo库当中正好就有定时器，因此我们就用muduo库的定时器了

2.LRU风格的信道置换

从信道池当中取出信道时，每次取出的都是最近最少使用的信道，也就是距离过期最近的信道

因此这里我们需要使用LRU算法思想，来进行信道置换

3.小总结

我们要实现的是一个基于LRU算法进行信道置换的，动态扩容，超时缩容的一个信道池get取信道时：取出最近最少使用的信道
eliminate淘汰信道时：根据超时机制淘汰信道
put新增信道时：新增信道，并且为该信道设置定时任务

关于LRU Cache，大家可以看我的博客：
LRU Cache
我们信道池当中的这个LRU算法是它的一个变形体

三、信道池 — “智能共享单车管理系统”

0.设计目标

通过动态调整信道【共享单车】数量，来适应线程【顾客】需求【调整供应以适应需求】

1.共享资源

信道跟共享单车一样，都属于公共资源，所有线程/人都可以申请使用，且使用完之后都需要归还

2.动态扩容

我们的这个智能共享单车管理系统是调整共享单车的供应来适应需求的

当可用共享单车数【信道数】为0时，我们会增加系统当中共享单车【信道】的数量

我们就按照
0 -> 4 -> 8 -> 16 -> 32这种方式来进行动态扩容

3.超时缩容

当共享单车【信道】位于我们的系统当中时，我们才会对他进行超时检测，当时间到了之后，若该信道依旧位于我们的系统当中，则淘汰该信道

因为此时供大于求，需要减少供应

4.返回最近最少使用的资源

当你想要从系统中拿取一份信道时，系统并不会直接给你最新放进来的信道，而是会智能地选择那份离超时最近的信道给你

这样做的好处是，能够最大限度地保持信道池中的信道处于活跃状态，同时减少因为长时间未使用而导致的资源浪费

5.LRU何在？

信道池在使用时采用LRU（最近最少使用）算法来返回信道，以确保信道池中的信道在符合需求时尽可能多地被利用，而超时淘汰机制则主要用于淘汰掉那些长时间未被使用【冗余】的信道。

这样做的好处是，能够最大限度地保持信道池中的信道处于活跃状态，同时减少因为长时间未使用而导致的资源浪费。

超时淘汰机制则是作为一个补充措施，用于进一步清理那些长时间未被使用或已经变得不再需要的信道。
通过设定合理的超时时间，系统可以自动地释放这些信道，避免资源浪费。这种机制有助于防止信道池中的资源被无限制地占用，从而保证了系统的稳定性和可用性。

它能够确保信道池中的资源得到充分利用，同时避免资源的浪费和系统的拥塞

四、muduo库的定时器介绍与使用

下面事不宜迟，带大家了解一下muduo库的定时器

1.接口

2.示例设计

我们要实现这么一个定时服务模块的示例：

对外提供一个set_timer接口，允许使用者注册定时任务
提供一个cancel_timer接口，允许使用者删除定时任务
提供一个shutdown接口，允许使用者删除所有定时任务，停止该服务

要求使用者构造该模块时传入一个TimerCallback 类型的回调函数，就是定时任务

对T类型的对象进行监控
template<class T>using TimerCallback = std::function<void(const std::shared_ptr<T>&)>;

我们要将  监控对象和TimerId  相绑定，而且需要快速查找，所以首先考虑使用unordered_map但是  并不是所有的类型都默认支持哈希，因此我们用shared_ptr把对象包起来
因为shared_ptr支持哈希【用shared_ptr底层对应的原生指针的地址作为唯一key值】所以我们将key值设置为std::shared_ptr<T>类型

3.代码实现

1.框架结构

template <class T>
class TimerHelper
{
public:using TimerCallback = std::function<void(const std::shared_ptr<T> &)>;TimerHelper(double delay,const TimerCallback &callback);void set_timer(const std::shared_ptr<T> &sp);void cancel_timer(const std::shared_ptr<T> &sp);~TimerHelper();void shutdown();private:double _delay;bool _running = true;TimerCallback _callback;// 信道池那里存在线程安全问题，因此要加互斥锁std::mutex _mutex;std::unordered_map<std::shared_ptr<T>, muduo::net::TimerId> timer_map;muduo::net::EventLoopThread loop_thread;muduo::net::EventLoop *loop;
};

2.构造和析构

注意：这里不能给delay缺省参数，因为缺省参数只能从右往左给

TimerHelper(double delay,const TimerCallback &callback): _delay(delay), _callback(callback), loop(loop_thread.startLoop()){}~TimerCallback()
{if (_running){shutdown();}
}

3.set_timer

其实就是向哈希表当中插入数据而已
只不过插入数据时要设置定时回调任务

因为我们的定时回调任务的函数签名是这样的：
using TimerCallback = std::function<void(const std::shared_ptr<T> &)>;而EventLoop设置回调函数的函数签名是这样的：
TimerId runEvery(double interval, TimerCallback cb);
typedef std::function<void()> TimerCallback;因此我们需要对_callback 给bind一下
loop->runEvery(_interval, std::bind(_callback, sp));返回的TimerId要保存到哈希表当中

所以set_timer就不难写出：

void set_timer(const std::shared_ptr<T> &sp)
{std::unique_lock<std::mutex> ulock(_mutex);auto iter = timer_map.find(sp);if (iter == timer_map.end()){timer_map[sp] = loop->runAfter(_delay, std::bind(_callback, sp));}else{std::cout << "该数据已经被设置过定时了\n";}
}

4.cancel_timer

EventLoop当中有cancel函数用来取消定时器
void cancel(TimerId timerId);删除定时器时先取消该定时器，然后在哈希表当中删除该定时器

void cancel_timer(const std::shared_ptr<T> &sp)
{std::unique_lock<std::mutex> ulock(_mutex);auto iter = timer_map.find(sp);if (iter != timer_map.end()){loop->cancel(iter->second);timer_map.erase(iter);}
}

5.完整代码

template <class T>
class TimerHelper
{
public:using TimerCallback = std::function<void(const std::shared_ptr<T> &)>;TimerHelper(double delay,const TimerCallback &callback): _delay(delay), _callback(callback), loop(loop_thread.startLoop()){}void set_timer(const std::shared_ptr<T> &sp){std::unique_lock<std::mutex> ulock(_mutex);auto iter = timer_map.find(sp);if (iter == timer_map.end()){timer_map[sp] = loop->runAfter(_delay, std::bind(_callback, sp));}else{std::cout << "该数据已经被设置过定时了\n";}}void cancel_timer(const std::shared_ptr<T> &sp){std::unique_lock<std::mutex> ulock(_mutex);auto iter = timer_map.find(sp);if (iter != timer_map.end()){loop->cancel(iter->second);timer_map.erase(iter);}}~TimerHelper(){if (_running){shutdown();}}void shutdown(){std::unique_lock<std::mutex> ulock(_mutex);if (_running){_running = false;for (auto &kv : timer_map){loop->cancel(kv.second);}loop->quit();}}private:double _delay;bool _running = true;TimerCallback _callback;// 存在多线程同时访问的需求，因此要加互斥锁std::mutex _mutex;std::unordered_map<std::shared_ptr<T>, muduo::net::TimerId> timer_map;muduo::net::EventLoopThread loop_thread;muduo::net::EventLoop *loop;
};

五、基于LRU置换模块的信道池实现

1.设计

借助上面实现的定时器模块完成我们的超时淘汰需求，下面我们要考虑的其实就是动态扩容和LRU的get，还有recover

1.LRU的get

从核心基于LRU的get入手，既然get需要返回最近最少使用的信道，势必就需要维护一个次序

而如何维护次序，我们就要想，次序何时会变

但是如果这样设计的话，被取走的信道我们就无法管理以及得知任何状态了。

我们作为信道池，且该信道池被用于我们这个项目当中的一个框架/组件，为了提高代码的可维护性和可读性，我们采用集中管理资源的方式

即：

1. 资源由专门的组件或模块管理
2. 资源的申请和释放通过统一的接口进行
3. 便于监控和管理资源的使用情况

因此我们需要管理被取走的信道，而我们的信道池是需要进行信道归还的，归还时就需要将对应信道从被取走信道集合当中删除
所以需要O(1)来查找对应信道

而且信道删除时也需要O(1)找到对应信道进行删除，因此需要一个unordered_map来提高查找效率

因此为了代码的优雅和统一性，我们再开一个list来存放被取走的信道

为了更好的标识所处状态，我们用枚举类型来标识状态enum State
{AVAILABLE,UNAVAILABLE
};class LRUChannelPool
{using iter_type = list<pair<Channel::ptr, State>>::iterator;// 链表： 头：最旧   尾：最新list<pair<Channel::ptr, State>> available_list;list<pair<Channel::ptr, State>> unavailable_list;unordered_map<Channel::ptr, iter_type> data_hash;
};

AVAILABLE：空闲的，可获得的

因此available_list当中的信道就是空闲信道，他们都需要进行定时任务，都有可能会超时淘汰
而unavailable_list当中的信道都是非空闲信道，他们都需要在未来被recovery后放回available_list当中

2.加锁的考量

是否需要加锁呢？

需要,因为我们的信道池是面向所有线程的,每个线程要想申请信道,必须且只能从信道池当中申请，所以我们的信道池需要考虑线程安全问题，
所以需要给个互斥锁

3.其他成员

double _delay;
ProtobufCodecPtr _codec;
muduo::net::TcpConnectionPtr _conn;
ChannelManager::ptr _channel_manager;std::shared_ptr<TimerHelper<Channel>> _helper;

4.接口

1. 构造
2. 析构（关闭所有信道即可），因为TimerHelper会调用自己的析构，所以不用管
3. get返回最近最少使用的信道
4. recover传入信道进行恢复
5. 检查并删除信道（给timer定时器绑定的回调函数）
6. 针对服务器的打开和关闭信道

2.实现

1.构造和析构

LRUChannelPool(const ProtobufCodecPtr &codec, const muduo::net::TcpConnectionPtr &conn, const ChannelManager::ptr &cp, double interval = default_delay) // 两分钟的超时时间: _delay(interval), _codec(codec), _conn(conn), _channel_manager(cp), _helper(std::make_shared<TimerHelper<Channel>>(_delay, std::bind(&LRUChannelPool::check_erase, this, std::placeholders::_1))){}

析构就是遍历两个链表中的所有信道进行关闭

~LRUChannelPool()
{std::unique_lock<std::mutex> ulock(_mutex);// 遍历两个链表中的所有信道进行关闭for (auto &kv : available_list){Channel::ptr cp = kv.first;CloseChannel(cp);}for (auto &kv : unavailable_list){Channel::ptr cp = kv.first;CloseChannel(cp);}
}

2.get获取信道

0. 加锁
1. 扩容
2. 取出available_list的队头元素,将其转移splice到unavailable_list当中【因为用splice可以直接转移节点，因此迭代器不会失效，无需更新哈希表】
3. 修改其State为UNAVAILABLE
4. 取消其定时
5. 将其返回

1.扩容

get的时候：当空闲信道数为0时，便会扩容。
扩完容之后，将available_list的容量扩大至_capacity-unavailable_list.size()

每次新增完信道，都要将该信道放入定时器进行监控

// 不能抽离出来expandActiveChannel进行复用，否则会造成线程安全问题（因为两个原子操作放在一起就不原子了）
std::unique_lock<std::mutex> ulock(_mutex);
// 这是扩容
if (available_list.size() == 0)
{int newCapacity = _capacity == 0 ? 4 : 2 * _capacity;_capacity = newCapacity;// 这是新增信道// 将active_list的容量扩为 _capacity-unavailable_list.size()int active_cap = _capacity - unavailable_list.size();while (available_list.size() < active_cap){Channel::ptr cp = OpenChannel();// 放入链表available_list.push_back({cp, AVAILABLE});// 放入哈希表data_hash.insert(std::make_pair(cp, std::prev(available_list.end())));// 添加定时器_helper->set_timer(cp);}
}

2.转移节点

这里就可以用list当中非常好用的splice了

iter_type iter_list = available_list.begin();
Channel::ptr cp = iter_list->first;
// 把available_list.begin()转移到unavailable_list链表的unavailable_list的尾部
unavailable_list.splice(unavailable_list.end(), available_list, iter_list);

因为avaliable_list的队头元素是最近最少使用的元素，而队尾是最新鲜的元素
因此恢复信道时是将对应信道从unavailable_list移动到available_list的尾部

3.修改状态，取消定时，并返回

// 转移之后将available改为unavailable
iter_list->second = UNAVAILABLE;// 取消该信道的定时
_helper->cancel_timer(cp);
default_info("获取信道成功: %s",cp->cid().c_str());
return cp;

4.完整代码

Channel::ptr get()
{// 不能抽离出来expandActiveChannel进行复用，否则会造成线程安全问题（因为两个原子操作放在一起就不原子了）std::unique_lock<std::mutex> ulock(_mutex);// 这是扩容if (available_list.size() == 0){int newCapacity = _capacity == 0 ? 4 : 2 * _capacity;_capacity = newCapacity;// 这是新增信道// 将active_list的容量扩为 _capacity-unavailable_list.size()int active_cap = _capacity - unavailable_list.size();while (available_list.size() < active_cap){Channel::ptr cp = OpenChannel();// 放入链表available_list.push_back({cp, AVAILABLE});// 放入哈希表data_hash.insert(std::make_pair(cp, std::prev(available_list.end())));// 添加定时器_helper->set_timer(cp);}}iter_type iter_list = available_list.begin();Channel::ptr cp = iter_list->first;// 把available_list.begin()转移到unavailable_list链表的unavailable_list的尾部unavailable_list.splice(unavailable_list.end(), available_list, iter_list);// 转移之后将available改为unavailableiter_list->second = UNAVAILABLE;// 取消该信道的定时_helper->cancel_timer(cp);default_info("获取信道成功: %s",cp->cid().c_str());return cp;
}

3.recover恢复信道

0. 加锁
1. 查找该信道
2. 根据State判断该信道在哪个链表当中并决定是否需要recover【处于available的信道无需恢复】
3. 取消该信道的消费者订阅
4. 转移节点
5. 修改State
6. 恢复定时

注意：因为一个信道只能关联一个消费者，而我们的信道池是复用信道的
因为每次我们要保证我们的信道池当中的信道都没有关联消费者，因此每次recover的时候都要取消一下信道

1.查找

std::unique_lock<std::mutex> ulock(_mutex);
// 将unavailable_list当中的指定信道移动到available_list的尾部【splice】
// 先查找该值在哈希表当中是否存在
auto iter_hash = data_hash.find(key);
if (iter_hash == data_hash.end())
{default_warning("恢复信道时，对应信道并未在哈希表当中找到");return;
}

2.根据State进行判断

iter_type iter_list = iter_hash->second;
// 为AVAILABLE，说明它在available_list，有问题：因为只有unavailable_list当中的信道才是非空闲信道，才有可能被恢复
//而available_list当中的信道是空闲信道，不能被恢复
if (iter_list->second == AVAILABLE)
{default_warning("恢复信道时，对应信道为空闲信道，不予恢复");return;
}

3.调用该信道的basicCancel来取消订阅

// 调用该信道的basicCancel来取消订阅
key->BasicCancel();

4.转移节点

// 移动该迭代器,放到空闲链表的尾部(保证最近最少使用的是在头部)
available_list.splice(available_list.end(), unavailable_list, iter_list);

5.修改State，恢复定时

// 移动之后,将他改为avaliable
iter_list->second = AVAILABLE;// 添加该信道的定时
_helper->set_timer(key);

6.完整代码

void recover(const Channel::ptr &key)
{std::unique_lock<std::mutex> ulock(_mutex);// 将unavailable_list当中的指定信道移动到available_list的尾部【splice】// 先查找该值在哈希表当中是否存在auto iter_hash = data_hash.find(key);if (iter_hash == data_hash.end()){default_warning("恢复信道时，对应信道并未在哈希表当中找到");return;}iter_type iter_list = iter_hash->second;// 为AVAILABLE，说明它在available_list，有问题：因为只有available_list当中的信道才是空闲信道，才有可能被恢复，不能恢复空闲信道if (iter_list->second == AVAILABLE){default_warning("恢复信道时，对应信道为空闲信道，不予恢复");return;}// 调用该信道的basicCancel来取消订阅key->BasicCancel();// 移动该迭代器,放到空闲链表的尾部(保证最近最少使用的是在头部)available_list.splice(available_list.end(), unavailable_list, iter_list);// 移动之后,将他改为avaliableiter_list->second = AVAILABLE;// 添加该信道的定时_helper->set_timer(key);default_warning("恢复信道成功 %s",key->cid().c_str());
}

4.检查并删除信道

0. 加锁
1. 在哈希表当中查找
2. 检查是否是空闲信道AVAILABLE（如果不是，则return）
3. 停止定时任务
4. 在available_list和哈希表当中删除该信道
5. 关闭该信道

因为第5步可以单独拎出来，无需加锁，所以为了减少锁冲突，我们将其提出来了

1.在哈希表当中查找

Channel::ptr cp;
{std::unique_lock<std::mutex> ulock(_mutex);// 先查找该值在哈希表当中是否存在auto iter_hash = data_hash.find(key);if (iter_hash == data_hash.end()){default_warning("check_erase信道时，对应信道并未在哈希表当中找到");return;}

2.检查是否空闲

iter_type iter_list = iter_hash->second;
// 为UNAVAILABLE，说明它在unavailable_list，有问题：因为只有available_list当中的信道才是空闲信道，才有可能被淘汰，不能淘汰非空闲信道
if (iter_list->second == UNAVAILABLE)
{return;
}

3.停止定时任务

cp = iter_list->first;
// 1. 停止对该信道的定时任务
_helper->cancel_timer(cp);

4.删除信道

// 2. 在available_list当中删除available_list.erase(iter_list);// 3. 在哈希表当中删除data_hash.erase(iter_hash);
}

5.关闭信道

// 拿到该信道，关闭并销毁，最后删除
CloseChannel(cp);

6.完整代码

void check_erase(const Channel::ptr &key)
{Channel::ptr cp;{std::unique_lock<std::mutex> ulock(_mutex);// 先查找该值在哈希表当中是否存在auto iter_hash = data_hash.find(key);if (iter_hash == data_hash.end()){default_warning("check_erase信道时，对应信道并未在哈希表当中找到");return;}iter_type iter_list = iter_hash->second;// 为UNAVAILABLE，说明它在unavailable_list，有问题：因为只有available_list当中的信道才是空闲信道，才有可能被淘汰，不能淘汰非空闲信道if (iter_list->second == UNAVAILABLE){return;}cp = iter_list->first;// 1. 停止对该信道的定时任务_helper->cancel_timer(cp);// 2. 在available_list当中删除available_list.erase(iter_list);// 3. 在哈希表当中删除data_hash.erase(iter_hash);}// 拿到该信道，关闭并销毁，最后删除CloseChannel(cp);default_warning("信道删除成功：%s",cp->cid().c_str());
}

5.打开与关闭信道

private:
Channel::ptr OpenChannel()
{// 1.创建channelChannel::ptr cp = _channel_manager->createChannel(_conn, _codec);// 2. 打开channelcp->openChannel();default_info("打开信道成功 %s",cp->cid().c_str());return cp;
}void CloseChannel(const Channel::ptr &cp)
{// 1. 关闭channelcp->closeChannel();// 2. 销毁channel_channel_manager->removeChannel(cp->cid());default_info("关闭信道成功 %s",cp->cid().c_str());
}

3.完整代码

enum State
{AVAILABLE,UNAVAILABLE
};const double default_delay = 1.0;// 存在多线程同时申请信道，因此信道池需要有互斥锁来保证线程安全
// 就像是学校食堂窗口，排队买饭的时候，如果饭准备较慢的话，都会给号码牌，那么很多人领号码牌的时候，要保证互斥+同步，保证线程安全
class LRUChannelPool
{
private:using iter_type = list<pair<Channel::ptr, State>>::iterator;public:using ptr = std::shared_ptr<LRUChannelPool>;LRUChannelPool(const ProtobufCodecPtr &codec, const muduo::net::TcpConnectionPtr &conn, const ChannelManager::ptr &cp, double interval = default_delay) // 两分钟的超时时间: _delay(interval), _codec(codec), _conn(conn), _channel_manager(cp), _helper(std::make_shared<TimerHelper<Channel>>(_delay, std::bind(&LRUChannelPool::check_erase, this, std::placeholders::_1))){}~LRUChannelPool(){std::unique_lock<std::mutex> ulock(_mutex);// 遍历两个链表中的所有信道，停止对该信道的定时任务，并且关闭信道for (auto &kv : available_list){Channel::ptr cp = kv.first;CloseChannel(cp);}for (auto &kv : unavailable_list){Channel::ptr cp = kv.first;CloseChannel(cp);}}Channel::ptr get(){// 不能抽离出来expandActiveChannel进行复用，否则会造成线程安全问题（因为两个原子操作放在一起就不原子了）std::unique_lock<std::mutex> ulock(_mutex);// 这是扩容if (available_list.size() == 0){int newCapacity = _capacity == 0 ? 4 : 2 * _capacity;_capacity = newCapacity;cout << "本次扩容后，新容量：" << _capacity << "，哈希表信道数：" << data_hash.size() << " 空闲信道数："<< available_list.size() << " ,非空闲信道数：" << unavailable_list.size() << "\n";// 这是新增信道// 将active_list的容量扩为 _capacity-unavailable_list.size()int active_cap = _capacity - unavailable_list.size();while (available_list.size() < active_cap){Channel::ptr cp = OpenChannel();// 放入链表available_list.push_back({cp, AVAILABLE});// 放入哈希表data_hash.insert(std::make_pair(cp, std::prev(available_list.end())));// 放入定时器队列_helper->set_timer(cp);}}iter_type iter_list = available_list.begin();Channel::ptr cp = iter_list->first;// 把available_list.begin()转移到unavailable_list链表的unavailable_list的尾部unavailable_list.splice(unavailable_list.end(), available_list, iter_list);// 转移之后将available改为unavailableiter_list->second = UNAVAILABLE;// 取消该信道的定时_helper->cancel_timer(cp);default_info("获取信道成功 %s",cp->cid().c_str());return cp;}void recover(const Channel::ptr &key){std::unique_lock<std::mutex> ulock(_mutex);// 将unavailable_list当中的指定信道移动到available_list的尾部【splice】// 先查找该值在哈希表当中是否存在auto iter_hash = data_hash.find(key);if (iter_hash == data_hash.end()){default_warning("恢复信道时，对应信道并未在哈希表当中找到");return;}iter_type iter_list = iter_hash->second;// 为AVAILABLE，说明它在available_list，有问题：因为只有available_list当中的信道才是空闲信道，才有可能被恢复，不能恢复空闲信道if (iter_list->second == AVAILABLE){default_warning("恢复信道时，对应信道为空闲信道，不予恢复");return;}// 移动该迭代器,放到空闲链表的尾部(保证最近最少使用的是在头部)available_list.splice(available_list.end(), unavailable_list, iter_list);// 移动之后,将他改为avaliableiter_list->second = AVAILABLE;// 添加该信道的定时_helper->set_timer(key);default_info("恢复信道成功 %s",key->cid().c_str());}private:void check_erase(const Channel::ptr &key){Channel::ptr cp;{std::unique_lock<std::mutex> ulock(_mutex);// 先查找该值在哈希表当中是否存在auto iter_hash = data_hash.find(key);if (iter_hash == data_hash.end()){default_warning("check_erase信道时，对应信道并未在哈希表当中找到");return;}iter_type iter_list = iter_hash->second;// 为UNAVAILABLE，说明它在unavailable_list，有问题：因为只有available_list当中的信道才是空闲信道，才有可能被淘汰，不能淘汰非空闲信道if (iter_list->second == UNAVAILABLE){return;}cp = iter_list->first;// 1. 停止对该信道的定时任务_helper->cancel_timer(cp);// 2. 在available_list当中删除available_list.erase(iter_list);// 3. 在哈希表当中删除data_hash.erase(iter_hash);}// 拿到该信道，关闭并销毁，最后删除CloseChannel(cp);default_info("信道删除成功：%s",cp->cid().c_str());}private:Channel::ptr OpenChannel(){// 1.创建channelChannel::ptr cp = _channel_manager->createChannel(_conn, _codec);// 2. 打开channelcp->openChannel();default_info("打开信道成功 %s",cp->cid().c_str());return cp;}void CloseChannel(const Channel::ptr &cp){// 1. 关闭channelcp->closeChannel();// 2. 销毁channel_channel_manager->removeChannel(cp->cid());default_info("关闭信道成功 %s",cp->cid().c_str());}private:std::mutex _mutex;// 链表： 头：最旧   尾：最新list<pair<Channel::ptr, State>> available_list;list<pair<Channel::ptr, State>> unavailable_list;unordered_map<Channel::ptr, iter_type> data_hash;int _capacity = 0;double _delay;ProtobufCodecPtr _codec;muduo::net::TcpConnectionPtr _conn;ChannelManager::ptr _channel_manager;std::shared_ptr<TimerHelper<Channel>> _helper;
};

六、修改Connection

1.修改部分

有了信道池这一中间层之后，且该中间层依然可以完成信道的打开与关闭，下面要做的就是把这个中间层加到连接模块与信道模块之间

因此在连接类当中需要加一个成员变量：

LRUChannelPool::ptr _channel_pool;

然后把原初的OpenChannel和CloseChannel改为获得与返回Channel
然后纯复用即可写出代码来：

Channel::ptr getChannel()
{if (_conn->connected() == false){default_fatal("获取信道失败,因为连接尚未建立");return Channel::ptr();}else if (_channel_pool.get() == nullptr){default_fatal("获取信道失败,因为信道池尚未初始化");return Channel::ptr();}return _channel_pool->get();
}void returnChannel(const Channel::ptr &cp)
{if (_conn->connected() == false){default_fatal("获取信道失败,因为连接尚未建立");return;}else if (_channel_pool.get() == nullptr){default_fatal("获取信道失败,因为信道池尚未初始化");return;}_channel_pool->recover(cp);
}

不过有一个问题：
Connection当中有一个成员函数：muduo::net::TcpConnectionPtr _conn;
它需要在OnConnectionCallback这个成员函数当中进行初始化
只有当他初始化了之后，他才是一个完整的TcpConnectionPtr ，否则他就是一个空的智能指针

如果在构造函数当中将其贸然传递给LRUChannel_pool的话，那么信道池当中的_conn就是空的智能指针了，就是天大的BUG

因此_channel_pool必须要在_conn初始化之后才能构造

又因为客户端在构造完Connection连接对象之后就会返回，然后线程就可以创建信道了

因此_channel_pool必须在connect函数调用结束之前就初始化

又因为OnConnectionCallback和connect是异步调用关系，是由两个线程分别调用的

一旦_latch.countDown();被调用，那么构造函数就会返回，所以_channel_pool必须在_latch.countDown();之前调用去初始化

因此，代码如下：

void connect()
{// 1. 客户端发起连接_client.connect();// 2. 等待连接建立成功_latch.wait();
}void OnConnectionCallback(const muduo::net::TcpConnectionPtr &conn)
{std::ostringstream oss;if (conn->connected()){_conn = conn;_channel_pool = std::make_shared<LRUChannelPool>(_codec, _conn, _channel_manager);// 必须先初始化信道池，后_latch.countDown放开connect过去_latch.countDown();default_info("连接建立成功");}else{_conn.reset();_channel_pool.reset();default_info("连接断开成功");}
}

2.总代码

#pragma once#include "muduo/net/TcpClient.h"
#include "muduo/net/TcpConnection.h"
#include "muduo/base/CountDownLatch.h"#include "proto/dispatcher.h"
#include "proto/codec.h"#include "channel.hpp"
#include "async_worker.hpp"#include "../mqcommon/mq_msg.pb.h"
#include "../mqcommon/mq_proto.pb.h"
#include "../mqhelper/logger.hpp"
#include "../mqhelper/helper.hpp"#include "channel_pool.hpp"namespace ns_mq
{namespace ns_google{using MessagePtr = std::shared_ptr<google::protobuf::Message>;}class Connection{public:using ptr = std::shared_ptr<Connection>;Connection(const std::string &server_ip, uint16_t server_port, const AsyncWorker::ptr &worker): _worker(worker), _latch(1), _client(_worker->_loopthread.startLoop(), muduo::net::InetAddress(server_ip, server_port), "Client"), _dispatcher(std::bind(&Connection::OnUnknownCallback, 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))), _channel_manager(std::make_shared<ChannelManager>())//,_channel_pool(std::make_shared<LRUChannelPool>(_codec,_conn,_channel_manager)){_dispatcher.registerMessageCallback<BasicCommonResponse>(std::bind(&Connection::OnCommonResponse, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_dispatcher.registerMessageCallback<BasicConsumeResponse>(std::bind(&Connection::OnConsumeResponse, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));_client.setConnectionCallback(std::bind(&Connection::OnConnectionCallback, this, std::placeholders::_1));_client.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(), std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));connect();}Channel::ptr getChannel(){if (_conn->connected() == false){default_fatal("获取信道失败,因为连接尚未建立");return Channel::ptr();}else if (_channel_pool.get() == nullptr){default_fatal("获取信道失败,因为信道池尚未初始化");return Channel::ptr();}return _channel_pool->get();}void returnChannel(const Channel::ptr &cp){if (_conn->connected() == false){default_fatal("恢复信道失败,因为连接尚未建立");return;}else if (_channel_pool.get() == nullptr){default_fatal("获取信道失败,因为信道池尚未初始化");return;}_channel_pool->recover(cp);}private:void connect(){// 1. 客户端发起连接_client.connect();// 2. 等待连接建立成功_latch.wait();}void OnUnknownCallback(const muduo::net::TcpConnectionPtr &conn, const ns_google::MessagePtr &message, muduo::Timestamp){default_info("未知请求, 我们将断开该连接");if (conn->connected()){conn->shutdown();}}void OnConnectionCallback(const muduo::net::TcpConnectionPtr &conn){if (conn->connected()){_conn = conn;_channel_pool = std::make_shared<LRUChannelPool>(_codec, _conn, _channel_manager);// 必须先初始化资源，后_latch.countDown放开connect过去_latch.countDown();default_info("连接建立成功");}else{_conn.reset();_channel_pool.reset();default_info("连接断开成功");}}void OnCommonResponse(const muduo::net::TcpConnectionPtr &conn, const BasicCommonResponsePtr &resp, muduo::Timestamp){// 找到该信道,然后将该响应添加到对应信道维护的相应哈希表当中Channel::ptr cp = _channel_manager->getChannel(resp->channel_id());if (cp.get() == nullptr){default_info("未找到该信道, 信道ID: %s",resp->channel_id().c_str());return;}cp->putResponse(resp);}void OnConsumeResponse(const muduo::net::TcpConnectionPtr &conn, const BasicConsumeResponsePtr &resp, muduo::Timestamp){// 1.找到信道Channel::ptr cp = _channel_manager->getChannel(resp->channel_id());if (cp.get() == nullptr){default_info("未找到该信道, 信道ID: %s",resp->channel_id().c_str());return;}// 2.将 调用该信道对应的consume任务包装一下抛入线程池_worker->_pool.put([cp, resp](){ cp->consume(resp); });}muduo::CountDownLatch _latch;AsyncWorker::ptr _worker;muduo::net::TcpClient _client;muduo::net::TcpConnectionPtr _conn;/*先构造TcpClient，然后通过它建立连接并获得TcpConnectionPtr，最后在不再需要这些连接时先销毁或重置TcpConnectionPtr，最后销毁TcpClient*/ProtobufDispatcher _dispatcher;ProtobufCodecPtr _codec;ChannelManager::ptr _channel_manager;LRUChannelPool::ptr _channel_pool;};
}

七、测试

1.多信道创建测试

1.代码

void sleep(int second)
{this_thread::sleep_for(chrono::seconds(second));
}int main()
{AsyncWorker::ptr worker = std::make_shared<AsyncWorker>();// 1. 创建连接和信道Connection::ptr conn = std::make_shared<Connection>("127.0.0.1", 8888, worker);Channel::ptr cp1 = conn->getChannel();Channel::ptr cp2 = conn->getChannel();Channel::ptr cp3 = conn->getChannel();Channel::ptr cp4 = conn->getChannel();Channel::ptr cp5 = conn->getChannel();sleep(3);conn->returnChannel(cp1);sleep(3);conn->returnChannel(cp2);sleep(3);conn->returnChannel(cp3);sleep(3);conn->returnChannel(cp4);sleep(3);conn->returnChannel(cp5);sleep(10);return 0;
}

2.演示

验证成功

2.多线程，多信道服务测试

1.代码

生产者：

#include "connection.hpp"
using namespace ns_mq;
#include <thread>
#include <vector>
using namespace std;// host1
void publisher1(const Connection::ptr &conn, const std::string &thread_name)
{// 1. 创建信道Channel::ptr cp = conn->getChannel();// 2. 创建虚拟机,交换机,队列,并进行绑定cp->declareVirtualHost("host1", "./host1/resource.db", "./host1/message");cp->declareExchange("host1", "exchange1", TOPIC, true, false, {});cp->declareMsgQueue("host1", "queue1", true, false, false, {});//cp->declareMsgQueue("host1", "queue2", true, false, false, {});cp->bind("host1", "exchange1", "queue1", "news.sport.#");//cp->bind("host1", "exchange1", "queue2", "news.*.zhangsan");// 3. 发送10条消息BasicProperities bp;bp.set_mode(DURABLE);bp.set_routing_key("news.sport.basketball");for (int i = 0; i < 10; i++){bp.set_msg_id(UUIDHelper::uuid());cp->BasicPublish("host1", "exchange1", &bp, "Hello -" + std::to_string(i));}// 4. 关闭信道conn->returnChannel(cp);
}// host2
void publisher2(const Connection::ptr &conn, const std::string &thread_name)
{// 1. 创建信道Channel::ptr cp = conn->getChannel();// 2. 创建虚拟机,交换机,队列,并进行绑定cp->declareVirtualHost("host2", "./host2/resource.db", "./host2/message");cp->declareExchange("host2", "exchange1", TOPIC, true, false, {});cp->declareMsgQueue("host2", "queue1", true, false, false, {});cp->declareMsgQueue("host2", "queue2", true, false, false, {});cp->bind("host2", "exchange1", "queue1", "news.sport.#");cp->bind("host2", "exchange1", "queue2", "news.*.zhangsan");// 3. 发送10条消息BasicProperities bp;bp.set_mode(DURABLE);bp.set_routing_key("news.sport.basketball");for (int i = 0; i < 10; i++){bp.set_msg_id(UUIDHelper::uuid());cp->BasicPublish("host2", "exchange1", &bp, "Hello -" + std::to_string(i));}// 4. 关闭信道conn->returnChannel(cp);
}int main()
{AsyncWorker::ptr worker = std::make_shared<AsyncWorker>();Connection::ptr myconn = std::make_shared<Connection>("127.0.0.1", 8888, worker);vector<thread> thread_v;thread_v.push_back(thread(publisher1, myconn, "thread1"));thread_v.push_back(thread(publisher2, myconn, "thread2"));for(auto& t:thread_v)t.join();return 0;
}

消费者：

#include "connection.hpp"
using namespace ns_mq;
#include <thread>
#include <vector>
#include <thread>
using namespace std;// 因为要拿到信道才能进行确认,所以这里需要把Channel::ptr bind过来
void Callback(const Channel::ptr &cp, const std::string &consumer_tag, const BasicProperities *bp, const std::string &body)
{// 1. 消费消息std::string id;if (bp != nullptr){id = bp->msg_id();}std::cout << consumer_tag << " 消费了消息: " << body << ", 消息ID: " << id << "\n";// 2. 确认消息if (bp != nullptr)std::cout << cp->BasicAck(id) << "\n";
}void consumer1(const Connection::ptr &conn, const std::string &thread_name)
{Channel::ptr cp = conn->getChannel();std::cout << "consumer1: 信道ID：" << cp->cid() << "\n";// 2. 创建虚拟机,交换机,队列,并进行绑定cp->declareVirtualHost("host1", "./host1/resource.db", "./host1/message");cp->declareExchange("host1", "exchange1", TOPIC, true, false, {});cp->declareMsgQueue("host1", "queue1", true, false, false, {});//cp->declareMsgQueue("host1", "queue2", true, false, false, {});cp->bind("host1", "exchange1", "queue1", "news.sport.#");//cp->bind("host1", "exchange1", "queue2", "news.*.zhangsan");// 3. 创建消费者cp->BasicConsume("host1", "consumer1", "queue1",std::bind(Callback, cp, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3), false);// 4. 等待消息while (true){std::this_thread::sleep_for(std::chrono::seconds(1000));}// 5. 关闭信道conn->returnChannel(cp);
}void consumer2(const Connection::ptr &conn, const std::string &thread_name)
{Channel::ptr cp = conn->getChannel();std::cout << "consumer2: 信道ID：" << cp->cid() << "\n";// 2. 创建虚拟机,交换机,队列,并进行绑定cp->declareVirtualHost("host2", "./host2/resource.db", "./host2/message");cp->declareExchange("host2", "exchange1", TOPIC, true, false, {});cp->declareMsgQueue("host2", "queue1", true, false, false, {});cp->declareMsgQueue("host2", "queue2", true, false, false, {});cp->bind("host2", "exchange1", "queue1", "news.sport.#");cp->bind("host2", "exchange1", "queue2", "news.*.zhangsan");// 3. 创建消费者cp->BasicConsume("host2", "consumer2", "queue1",std::bind(Callback, cp, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3), false);// 4. 等待消息while (true){std::this_thread::sleep_for(std::chrono::seconds(1000));}// 5. 关闭信道conn->returnChannel(cp);
}void sleep(int second)
{this_thread::sleep_for(chrono::seconds(second));
}int main()
{AsyncWorker::ptr worker = std::make_shared<AsyncWorker>();// 1. 创建连接和信道Connection::ptr conn = std::make_shared<Connection>("127.0.0.1", 8888, worker);vector<thread> thread_v;thread_v.push_back(thread(consumer1, conn, "thread1"));thread_v.push_back(thread(consumer2, conn, "thread2"));for (auto &t : thread_v)t.join();return 0;
}

2.演示

验证成功

以上就是项目扩展一：信道池的实现的全部内容