Redis做消息队列

要使用Redis实现消息队列，需要了解消息队列都要满足哪些特性，消息队列在存取消息时，必须要满足消息的消费顺序不能乱，不能重复处理同一个消息，不能漏发消息，对应的我们可以分别称其为消息保序、消息幂等处理和消息可靠性保证。

基于 List 的消息队列解决方案

List 本身就是按先进先出的顺序对数据进行存取的，所以，如果使用 List 作为消息队列保存消息的话，就已经能满足消息保序的需求了，具体来说就是一个push，pop的操作，如下图所示：

image.png

这里生产者有消息写入时，不会通知消费者，为了解决这个问题，Redis 提供了 BRPOP 命令。BRPOP 命令也称为阻塞式读取，客户端在没有读到队列数据时，自动阻塞，直到有新的数据写入队列，再开始读取新数据。
第一个保序的问题解决了，还有重复消息处理的问题，这里消息队列要能给每一个消息提供全局唯一的 ID 号；另一方面，消费者程序要把已经处理过的消息的 ID 号记录下来。
这里消息的全局唯一 ID 号就需要生产者程序在发送消息前自行生成。

LPUSH mq "101030001:stock:5"      //将ID为101030001，库存量为5的消息插入消息队列
(integer) 1

最后，List 类型是如何保证消息可靠性的呢，List 类型提供了 BRPOPLPUSH 命令，这个命令的作用是让消费者程序从一个 List 中读取消息，同时，Redis 会把这个消息再插入到另一个 List（可以叫作备份 List）留存。
这样一来，如果消费者程序读了消息但没能正常处理，等它重启后，就可以从备份 List 中重新读取消息并进行处理了。如下图所示：

image.png

基于 Streams 的消息队列解决方案

使用List做消息队列，我们还可能遇到过一个问题，生产者消息发送很快，而消费者处理消息的速度比较慢，这就导致 List 中的消息越积越多，给 Redis 的内存带来很大压力。
Streams 是 Redis 专门为消息队列设计的数据类型，它提供了丰富的消息队列操作命令。

XADD：插入消息，保证有序，可以自动生成全局唯一 ID；
XREAD：用于读取消息，可以按 ID 读取数据；
XREADGROUP：按消费组形式读取消息；
XPENDING 和 XACK：XPENDING 命令可以用来查询每个消费组内所有消费者已读取但尚未确认的消息，而 XACK 命令用于向消息队列确认消息处理已完成。

XADD 命令可以往消息队列中插入新消息，消息的格式是键 - 值对形式。

XADD mqstream * repo 5   //往名称为mqstream的消息队列中插入一条消息，消息键为repo，值为5，带*的标识表示为这个消息队列自动生成一个全局唯一ID
"1599203861727-0"     // 全局唯一ID，前面的1599203861727标识生成时间，后面的0表示1599203861727毫秒这个时间的第一条消息

XREAD 在读取消息时，可以指定一个消息 ID，并从这个消息 ID 的下一条消息开始进行读取。

XREAD BLOCK 100 STREAMS  mqstream 1599203861727-0    //这里设置了block参数，100表示阻塞时长，这里表示读取1599203861727-0后续消息，一共读取了三条
1) 1) "mqstream"
   2) 1) 1) "1599274912765-0"
         2) 1) "repo"
            2) "3"
      2) 1) "1599274925823-0"
         2) 1) "repo"
            2) "2"
      3) 1) "1599274927910-0"
         2) 1) "repo"
            2) "1"

Streams 本身可以使用 XGROUP 创建消费组，创建消费组之后，Streams 可以使用 XREADGROUP 命令让消费组内的消费者读取消息。

XGROUP create mqstream group1 0    //创建一个名为group1的消费组，消费组中有消息队列mqstream
OK

//让group1中的消费者consumer1从mastream中读取所有消息，参数“>”，表示从第一条尚未被消费的消息开始读取

XREADGROUP group group1 consumer1 streams mqstream >
1) 1) "mqstream"
   2) 1) 1) "1599203861727-0"
         2) 1) "repo"
            2) "5"
      2) 1) "1599274912765-0"
         2) 1) "repo"
            2) "3"
      3) 1) "1599274925823-0"
         2) 1) "repo"
            2) "2"
      4) 1) "1599274927910-0"
         2) 1) "repo"
            2) "1"

使用消费组的目的是让组内的多个消费者共同分担读取消息，所以，我们通常会让每个消费者读取部分消息，从而实现消息读取负载在多个消费者间是均衡分布的。
Streams 会自动使用内部队列（也称为 PENDING List）留存消费组里每个消费者读取的消息，直到消费者使用 XACK 命令通知 Streams“消息已经处理完成”。

XPENDING mqstream group2    //查看group2中已经读取但是还没确认的消息个数，这里是3个
1) (integer) 3
2) "1599203861727-0"    //读取消息的最小ID
3) "1599274925823-0"    //读取消息的最大ID
4) 1) 1) "consumer1"
      2) "1"
   2) 1) "consumer2"
      2) "1"
   3) 1) "consumer3"
      2) "1"

一旦消息 1599274912765-0 被 consumer2 处理了，consumer2 就可以使用 XACK 命令通知 Streams，然后这条消息就会被删除。

 XACK mqstream group2 1599274912765-0
(integer) 1
XPENDING mqstream group2 - + 10 consumer2
(empty list or set)

最后我们总结一下基于List和基于Streams实现消息队列的特点，如下图所示：

image.png

这里要注意，Streams是Redis5.0的版本加入的，也就是要在5.0以上的版本才能使用它。