Redis中如何实现消息队列和延时消息队列

本文将详细解释如何在Redis中实现消息队列和延迟消息队列。边肖觉得挺实用的，分享给大家参考。希望你看完这篇文章能有所收获。

list的几个命令

左推

从队列的左侧存储它。

右推

从队列的右边开始存储。

左波普

从队伍的左边拿出来

rpop(右弹出)

从队伍的右边拿出来

以上四个命令可以使列表帮助我们实现队列或堆栈。队列以先进先出为特征，堆栈以先进先出为特征。

因此，队列的实现可以使用lpush rpop或rpush lpop，

栈的实现是lpush lpop或rpush rpop。

使用命令演示队列

生产者发布消息

首先，我们使用rpush向一个名为notify-queue的队列添加五个元素，即1 2 3 4 5，也就是作为生产者发布消息。

消费者消费消息

由于生产者使用rpush，那么消费者将使用lpop。可以看到下图。我们继续使用notify队列中的消息，它们从1到5依次被读出。最后，队列中没有消息，弹出的消息总是空的。

00-1010当使用上面的lpop消费消息时，我们可以看到在消息被消费之后，每次我们去pop的时候，我们都会读到一条空消息。

上面的命令是手动执行的，但是如果编写的代码程序一直去弹出数据(拉数据)，就会造成空轮询(无用读取)。

不仅提高了客户端的CPU消耗，还提高了redis的QPS，仍然是无用的操作，可能会导致其他客户端对redis的访问响应变慢。

解决方案A (休眠)

由于空轮询会使客户端和redis双方的资源消耗更高，我们可以让客户端在接收到空数据时休眠1s，然后在1s后拉取数据，这样可以降低消耗。

Thread.sleep(1000)

这个方案还有一个缺陷，就是增加了消息消耗的延迟。如果只有一个消费方，延迟为1s，也就是空轮询后，它只是进入睡眠状态，但此时刚好有消息传来，仍然需要等待1s醒来后再消费。

如果有多个消费者，因为每个消费者的睡眠时间不同，会减少一些延迟，但是有没有更好的方法实现几乎零延迟呢？

解决方案B (阻塞读)

关于redis中的队列获取，实际上有两个命令，即阻塞读取，

阻挡左击

brpop(阻止右弹出)

当队列中没有数据时，被阻止的读取将进入睡眠状态。一旦消息到达，它将立即响应并读取数据。因此，用blpop/brpop代替lpop/rpop可以解决消息延迟的问题。

继续排队3个属性，6，7和8。

使用blpop读取队列。最后一个参数是阻塞读取的等待时间。如果这次之后没有消息，将返回零。此时，您可以继续重复blpop操作。

阻塞读的空闲连接自动断开问题

客户端使用阻塞读取时，如果阻塞时间过长，服务一般会将其视为空闲连接，从而主动断开，减少无用连接占用的资源。此时，客户端将抛出一个异常。

因此，请注意，当客户端使用阻塞读取时，需要捕捉异常并进行相应的处理，如重试。

java客户端实现消息队列

ng>

思路和上述一样，只是由命令行客户端redis-cli变成了java语言，一个线程或多个线程进行 rpush 的发布，

另外一个或多个线程进行 blpop 消费，完成的代码在：https://github.com/qiaomengnan16/redis-demo/tree/main/redis-queue

发布者

订阅者

延时队列的实现思路

延时队列指的是，消息发送的一段时间后，再由消费者进行消费，而不是发送过去后，消费者就能立即读取到，

zset的可以帮我们做到这个事情，首先zset可以通过score进行排序，score可以存一个时间戳，所以我们每次发布消息的时候，用当前时间戳加上延时的时间戳，

随后消费者取消息的时候，通过截取zset的数据，取到已经满足当前时间的消息（即取score小于等于当前时间戳的数据，score小于等于当前时间戳代表消息已经到时间了，如果大于的话，说明还得等一会才能消费）。

关键命令 zadd (发布者)，zrangebyscore（订阅者），zrem (订阅者消费完数据后删除)

命令实现

我们使用zadd添加了4个数据，分别是1、2、3秒(伪说法，这里其实只是个score)后才能消费的数据，还有一个10秒后才能消费的kafka，

假如现在已经到了第三秒，我们取zset中大于等于1秒的和小于等于3秒的数据，因为这个区间的数据正好是我们可以消费的，可以看到，我们取出了符合条件的3条数据，

如果每次只能消费一个数据的话，可以加一个limit限制条件，可以看下图取出了第一个可以消费的数据，redis

同时注意，和list的lpop/和blpop不同（它们弹出即自动删除原始队列里的该数据），

虽然获取到了数据，但是如果不使用zrem进行删除的话，这条数据还会被其他人读到，因为他还一直存在zset中，

不过zrem可能会发生已经被别人抢先一步删除(消费)的情况，所以代码中还需要根据zrem的返回值是否大于0判断，本次消息我们是否抢占成功，成功后再进行正确消费。

代码实现

发布者

订阅者

测试延迟效果

完整代码地址：https://github.com/qiaomengnan16/redis-demo/tree/main/redis-delayed-queue

优化, 使用lua实现

上面实现的延迟队列中，有一个问题，就是使用zrem判断是否抢到这个数据时，很有可能没有抢到，这样继续进行读取，可能几轮都抢不到，资源白白浪费了，所以可以通过lua脚本，来进行优化，

让zrangebyscore 和 zrem成为一个原子化操作，这就可以避免多线程争抢，抢不到的资源浪费了。

关于“Redis中如何实现消息队列和延时消息队列”这篇文章就分享到这里了，希望