剖析 Redis List 消息队列的三种消费线程模型
Redis 列表（List）是一种简单的字符串列表，它的底层实现是一个双向链表。
生产环境，很多公司都将 Redis 列表应用于轻量级消息队列 。这篇文章，我们聊聊如何使用 List 命令实现消息队列的功能以及剖析消费者线程模型 。
1 核心流程
生产者使用 LPUSH key element[element...] 将消息插入到队列的头部，如果 key 不存在则会创建一个空的队列再插入消息。
如下，生产者向队列 queue 先后插入了 「Java」「勇哥」「Go」，返回值表示消息插入队列后的个数。
消费者使用 RPOP key 依次读取队列的消息，先进先出，所以 「Java」会先读取消费：
接下来，我们可以通过 spring-data-redis API 演示生产消费流程：
生产者
消费者
我们启动一个独立的线程从队列中读取消息（RPOP 命令），读取成功之后，消费消息，若没有消息，则休眠一会，下一次循环再继续。
上图的伪代码中， while(true) 循环内不停地调用 RPOP 指令，当有消息时，可以及时处理，但假如没有读取到消息，则需要休眠一会。
这里要加休眠，主要是为了减少空读的频率，避免 CPU 无意义的消耗。
有什么更优化的方式吗？ 有，那就是使用 Redis 阻塞读取 List 的命令。
Redis 提供了 BLPOP、BRPOP 阻塞读取的命令，消费者在在读取队列没有数据的时自动阻塞，直到有新的消息写入队列，才会继续读取新消息执行业务逻辑。
参数 0 表示阻塞等待时间无限制 。
如图，我们启动一个消费线程永动机，消费线程拉取消息后，执行消费逻辑。
这种消费者线程模型非常容易理解，同时也非常适合顺序消费的模式。同时，假如我们在消费消息时，服务器宕机或者断电，可能丢失一条消息。
接下来，我们想一想，有没有消费速度更高的消费模型吗？ 笔者根据过往的经历，列举三种模式：
拉取线程 + 消费线程池（非阻塞模式）
拉取线程 + 消费线程池 （阻塞模式）
拉取线程 + Disruptor（阻塞模式）
2 拉取线程 + 消费线程池（非阻塞模式）
为了提升消费速度，我们可以将拉取和消费拆分成两种动作，分别通过不同的线程池来处理。拉取线程池负责拉取消息，消费线程池负责消费消息。
伪代码类似：
如图，在拉取线程内部，我们拉取完消息后，将消息提交到消费线程 consumeExecutor 。
这样方式可以通过多线程执行大幅度提升消费速度 ，但是这里还是有一个问题：
假如消费速度很慢，生产者速度很高，那么就会在线程池内容易产生消息堆积，这里面会产生两个隐形风险：
线程池队列无限堆积，则可能有 OOM 的风险 ；
假如消费者服务器宕机或者断电，那么会丢失大量的消息。
那么如何优化这种模式呢 ？
答案是：拉取线程提交消息到线程池时，当队列中消息数量到达一定数量时，提交消息到线程池会阻塞。
3 拉取线程 + 消费线程池（阻塞模式）
我们将消息包装为 Runnable ，然后通过消费线程池执行 execute ，拉取线程会不会阻塞呢 ？
下图是执行的源码：
可以看到，第 30 行调用的是 workQueue 的非阻塞的 offer 方法。
如果队列已满，新提交的任务并不会被 block 住，反而会调用后续的 reject 流程。
如果我们想要达到阻塞生产者的目的的话，可以采取如下的两种方案：
信号量限制同时进入线程池等待队列的任务数 。
使用线程池的拒绝机制，把新加入的任务 put 到等待队列里，这样也可以阻塞住生产者。
4 拉取线程 + Disruptor
下图展示了 Disruptor 的流程图 。
和线程池机制非常类似， Disruptor 也是非常典型的生产者/消费者模式。线程池存储提交任务的容器是阻塞队列，而 Disruptor 使用的是环形缓冲区 RingBuffer。
环形缓冲区的设计相比阻塞队列有如下优点：
环形数组结构
为了避免垃圾回收，采用数组而非链表。同时，数组对处理器的缓存机制更加友好。
元素位置定位
数组长度 2^n，通过位运算，加快定位的速度。下标采取递增的形式，不用担心 index 溢出的问题。index 是 long 类型，即使 100 万 QPS 的处理速度，也需要 30 万年才能用完。
无锁设计
每个生产者或者消费者线程，会先申请可以操作的元素在数组中的位置，申请到之后，直接在该位置写入或者读取数据。
此刻大家并不需要理解环形缓冲区的读写机制，只需要明白 环形缓冲区 RingBuffer 是 Disruptor 的精髓即可。
将消费线程池替换成 Disruptor 有两个明显的优点：
无锁队列，写入读取性能非常好
当拉取线程提交消息到 Disruptor 时，若环形缓冲区 RingBuffer 已经满了，则拉取线程会阻塞，这样天然的可以避免无限拉取，同时避免 OOM 的问题。
伪代码类似：
1、定义 Disruptor
2、拉取线程将消息发送到 Disruptor Ringbuffer
3、消费消息
整体的消费者线程模型如下图：
5 平滑停服 + 定时任务补偿
当我们分析消费者线程模型时，无论我们使用哪种方式，假如服务器突然宕机、或者物理机断电，则会丢失消息。
笔者推荐两种方式：
1、平滑停服
平滑停服是指在停止应用程序时，尽量避免中断正在进行的请求或任务，尽量让正在进行的任务处理完成，并且不再接收新的任务，等所有任务执行完成后关闭应用。
在 Unix/Linux 系统中，可以使用 kill 命令发送信号给运行中的进程。
常见的信号有：
SIGTERM(15)：请求进程终止，可以被捕捉和处理，用于优雅地停止进程。SIGKILL(9)：强制终止进程，不能被捕捉或忽略。SIGQUIT(3)：进程退出并生成核心转储（core dump）。
为了实现平滑停服，可以使用 Java 的 Runtime.getRuntime().addShutdownHook 方法注册一个关闭钩子（shutdown hook）。当 JVM 接收到SIGTERM信号时，关闭钩子会被执行，从而可以在应用程序停止前执行一些清理工作。
我们可以在钩子里，关闭拉取线程池 ，优雅关闭消费线程池等 ，这样可以尽量避免丢失消息。
2、定时任务补偿
使用 List 做消息队列，不可避免的会有消息丢失，所以我们需要用定时任务做补偿，每隔一段时间去业务表里查询业务状态机，若状态机不符合条件，则触发补偿策略。
文章转载自：勇哥编程游记
还未添加个人签名 2023-06-19 加入
还未添加个人简介
评论