该篇主要是介绍下,使用Redis 中消息队列的四种实现方式:
使用 List 实现消息队列
rpush生产消息,lpop消费消息。当lpop没有消息的时候,要适当sleep一会再重试。
(缺点:使用 List 同样存在一定的问题,比如消息不支持重复消费、没有按照主题订阅的功能、不支持消费消息确认等)
如果对方追问可不可以不用sleep呢?list还有个指令叫blpop,在没有消息的时候,它会阻塞住直到消息到来。
使用 ZSet 实现消息队列
它是利用 zadd 和 zrangebyscore 来实现存入和读取消息的,这里就不重复叙述了。但 ZSet 的实现方式更为复杂一些,因为 ZSet 多了一个分值(score)属性,我们可以使用它来实现更多的功能,比如用它来存储时间戳,以此来实现延迟消息队列等。
ZSet 同样具备持久化的功能,List 存在的问题它也同样存在,不但如此,使用 ZSet 还不能存储相同元素的值。因为它是有序集合,有序集合的存储元素值是不能重复的,但分值可以重复,也就是说当消息值重复时,只能存储一条信息在 ZSet 中。
使用pub/sub主题订阅者模式,可以实现1:N的消息队列
缺点:
- 无法持久化保存消息,如果 Redis 服务器宕机或重启,那么所有的消息将会丢失;
- 发布订阅模式是“发后既忘”的工作模式,如果有订阅者离线重连之后就不能消费之前的历史消息;
- 不支持消费者确认机制,稳定性不能得到保证,例如当消费者获取到消息之后,还没来得及执行就宕机了。因为没有消费者确认机制,Redis 就会误以为消费者已经执行了,因此就不会重复发送未被正常消费的消息了,这样整体的 Redis 稳定性就被没有办法得到保障了。
使用 Stream 实现消息队列
Redis 5.0 之后新增了 Stream 类型,我们就可以使用 Stream 的 xadd 和 xrange 来实现消息的存入和读取了,并且 Stream 提供了 xack 手动确认消息消费的命令,用它我们就可以实现消费者确认的功能了,使用命令如下:
xack mq group1 1580959593553-0
相关语法如下:
xack key group-key ID [ID ...]
消费确认增加了消息的可靠性,一般在业务处理完成之后,需要执行 ack 确认消息已经被消费完成,整个流程的执行如下图所示:
实现延时队列
使用sortedset,拿时间戳作为score,消息内容作为key调用zadd来生产消息,消费者用zrangebyscore指令获取N秒之前的数据轮询进行处理。
到这里,面试官暗地里已经对你竖起了大拇指。但是他不知道的是此刻你却竖起了中指,在椅子背后。