队列的实现

举例：
队列主要用在系统解耦、流量削峰、异步处理、数据顺序处理等场景。新手在使用时可能会犯一些常见的错误。下面讲一个新手容易犯的错误，在这个示例中把队列的入队、出队和Redis存储节点的主从关系给混淆了，示例如下

存储：
Redis主节点M, 使用数据List类型做为队列,列表名称M (标记为M.L，意为主节点上的List)
Redis从节点S, 做为Redis主节点的备，有和主节点相同的列表名称L (标记为S.L，意为从节点上的List

应用程序：
应用名称App

操作：
入队命令： LPUSH M.L Data
出队命令： RPOP S.L

我们知道，Redis主从节点数据的流向是主节点－>从节点，队列中一般数据也是队尾（入队）－> 队头（出队），这两个数据流向混淆后就会出现以上的错误。

image.png

上图显然不是我们想要的结果，这种设计导致的问题是Redis主节点使用的内存会不断增长直至触发Redis的LRU策略导致数据丢失或者无法入队。Redis的从节点S仅仅是做为主节点M的一个备份节点存在的。

image.png

阻塞队列

阻塞队列是一种特殊的队列，具体是指对出队动作在队列为空时的阻塞行为以及在有元素入队后对出队的通知行为.我们知道事件通知机制是服务端通过一定的途径向客户端发送事件消息来实现的，服务端和客户端通常保持长链接。生产者（通知方）向队列中发送事件消息，消费者（接收方）从队列中拿走（POP）事件消息，当队列中没有事件消息的时候，消费者（接收方）阻塞，消费者（接收方）和队列之间保持长链接。

消费者（接收方）的代码（Java）逻辑应该大致如下:

    while (true){
            try {
                EventMessage em = client.brpop(list, timeout);
                doSomething(em);
            }catch(Exception e){
                continue;
            }
      }

使用循环是因为每次brpop命令只能拿到一个队列中的元素，为了得到消息的连续性必须循环brpop。在循环中使用try...catch...是防止在brpop的过程中由于网络闪断、连接池等因素导致的连接不可用抛出异常致使循环不可继续不能连续获取事件消息。

生产者（通知方）的Redis命令逻辑大致如下：

MULTI
      some command...
      LPUSH list x
      EXEC

阻塞队列与PUB/SUB在实现事件通知上的区别

• 1.PUB/SUB方式在消息消费者（被通知方）订阅一个或多个话题后和Redis保持长链接等待通知，之后可以连续的得到事件通知消息；阻塞队列brpop/blpop方式每次只能获取一个事件消息（即便pop多个队列也是一样），客户端需要循环使用brpop/blpop来获取事件消息，直至队列为空阻塞。
• 2.PUB/SUB方式实现的通知只有在消息消费者（被通知方）和Redis服务有在线连接的情况下才能收到，一旦连接断开消费者（被通知方）将不能收到通知，即便重新成功连接也无法收取丢失的通知消息；而阻塞队列在客户端断开后重新连接成功后可以收到通知消息。
• 3.PUB/SUB方式事件消息消费者可以有多个并且每个消费者都能得到相同的消息；而阻塞队列虽然事件消费者可以有多个但是消息只是分发给其中一个消费者，消息无法重复消费。

可靠队列

在Redis的列表(List)实现的队列中，一般一个客户端通过LPUSH命令将消息放入队列中，而另一个客户端通过RPOP/BRPOP 命令有顺序的取出队列中的消息进行消费。但是这种方式却不一定安全，因为在这个过程中，一个客户端可能在取出一个消息之后在处理这个消息之前崩溃，而未处理完的消息也就因此而丢失，并且无法找回。在一些希望是更可靠的消息传递系统中的应用上，这可能会导致一些问题。使用 RPOPLPUSH/BRPOPLPUSH 命令可以解决这个问题：因为它不仅返回一个消息，同时还将这个消息添加到另一个备份列表当中，当一个客户端完成某个消息的处理之后，可以从备份表中把消息删除。

image.png

对于备份列表一般有两种设计方式：每个客户端各自私有一个备份列表和所有客户端公用一个备份列表。两种方式实现机制不同，举例如下

• 1. 每个客户端各自私有一个备份队列
• 客户端从队列里获取消息之前首先检查自己的私有备份队列。
• 如果备份队列中有消息说明这条消息没有处理完，使用RPOPLPUSH/BRPOPLPUSH把这个消息取出来的同时再次放入备份队列（再次放入是防止从备份队列中取出后没处理完丢失），处理消息完成后把备份队列中的消息删除。
• 如果备份队列中没有消息，从消息队列中使用RPOPLPUSH/BRPOPLPUSH获取消息并把消息放入备份队列，处理完消息后把备份队列中的消息删除。

按照规则形成的处理逻辑，对每一个客户端分别使用自己私有的备份队列从根本上规避了所有客户端公用备份队列中由于“后发先至”等问题导致的复杂逻辑。

image.png

• 2.所有客户端公用一个备份队列，专门客户端做确认检查

• 所有客户端从队列中取出消息数据并通过RPOPLPUSH/BRPOPLPUSH放入备份队列

• 所有客户端处理消息完成后再从队列中取出下一个消息重复以上动作

• 专门客户端处理备份消息队列校对消息是否做完，如果做完则删除，否则重做消息。专门客户端起到一个补漏的作用。

image.png

旋转队列

在使用RPOPLPUSH命令的时候，它的两个参数如果是相同的队列键，客户端就可以一个接一个的获取从队头到队尾的所有元素并且把获取的元素放置到队尾。我们称作队列的旋转。这种情况在多个客户端对同一个队列进行旋转时也有效。甚至有客户端在旋转时增加队列的元素也是可行的。

此种模型对于队列和客户端来说都是易于扩展的。因为即便客户端获取元素后失败旋转到下一个周期后仍能获取到元素因此是安全的。这个模型使得我们可以很容易实现这样一类系统：有多个客户端，需要连续不断地对一些元素进行周期性的遍历轮训处理。一个典型的例子就是数据定时采集系统：采集点保存在队列中，客户端从队列中拿采集点然后采集数据。

image.png

优先权队列

优先队列和一般队列有所不同，它不完全遵循先进先出的规则，而是根据队列中元素的优先权，优先权最大的先被取出。队列中的元素不仅存储元素本身还存储元素的优先权数据。使用Redis数据结构实现的方式是按照优先权建队列（列表），相同优先权的元素在同一个队列中，客户端在使用BRPOP/RPOP命令使队列中的元素出队的时候参数按照优先权从高到低的顺序进行。

例如，

• 队列1：包含0～9，10个元素
• 队列2：包含0～3，4个元素
• 队列3：包含0～5，6个元素

优先权：队列1>队列2>队列3
客户端使用命令为：RPOP/BRPOP 队列1 队列2 队列3

image.png

个人介绍：

高广超：多年一线互联网研发与架构设计经验，擅长设计与落地高可用、高性能、可扩展的互联网架构。

本文首发在 高广超的简书博客 转载请注明！

简书博客

头条号