Redis 实现消息队列

71 阅读 0 评论 47 点赞

我是靠谱客的博主爱笑钥匙，最近开发中收集的这篇文章主要介绍Redis 实现消息队列，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

引子

面试的时候你提到了，Redis 和 MQ，面试官可能会让你用 Redis 实现消息队列，一方面考察你 Redis 的掌握，又考察了你对 MQ 的理解。可谓 一箭双雕。

消息队列

消息队列在分布式系统中用途非常广泛。
它具有 低耦合、可靠投递、广播、流量控制、最终一致性 等一系列功能。
常见的 消息队列 有 RabbitMQ、RocketMQ、ActiveMQ、Kafka

消息队列是指利用高效可靠的消息传递机制进行与平台无关的数据交流，并基于数据通信来进行分布式系统的集成。
通过提供消息传递和消息排队模型，它可以在分布式环境下提供 应用解耦、弹性伸缩、冗余存储、流量削峰、异步通信、数据同步 等等功能，其作为分布式系统架构中的一个重要组件，有着举足轻重的地位。

三个角色：生产者、消费者、消息中心
处理流程：生产者生产消息发送到消息中心，无需等待返回，消费者监听或者订阅此消息，获取消息。此过程，两者互不干扰，异步、松耦合。
可靠性：消息要可以保证不丢失、不重复消费、有时可能还需要顺序性的保证

Redis 实现消息队列

这个问题是我真实面试遇到过的问题，下面展示的是我当时回答的原话。

1.生产者事务很好解决，自带api就解决了 lpush rpush
2.redis持久化策略保证消息不丢失 AOF + RDB
3.主要是消费者这里，首先A组消费者和 B组消费者都可以去消费，那么他们组内是不可重复消费的。
这里每个组我可以创建一个key，value初始为-1（代表在队列里面的下标，其实就是 offset），可以increment成功之后（当然首先要判断offset要小于 List.length），根据这个值去队列里面取对应的消息。当然目前是无法保证事务的，保证事务的话，还需要一把锁，一个组只有一个消费者去消费，消费失败，offset -1 。

其实，是借用了 Kafka 和 RocketMq 的一些思想，大家 兼听则明 。

List 实现消息队列

LPUSH、RPOP 左进右出
RPUSH、LPOP 右进左出

127.0.0.1:6379> lpush queue a b c
(integer) 3
127.0.0.1:6379> rpop queue
"a"
127.0.0.1:6379> rpop queue
"b"
127.0.0.1:6379> rpop queue
"c"

简单的生产者-消费者模式就出来了。
要想 即时消费，比如说有消息了，消费者立刻拿到消息，然后返回。可以使用 BLPOP、BRPOP 这种阻塞式读取的命令。

LPUSH、BRPOP 左进右阻塞出
RPUSH、BLPOP 右进左阻塞出

目前这种，消息弹出，如果消费者异常（即消费失败），也无法回滚。为解决这种问题，可以引入 ACK 机制，其实类似于我当时面试时，使用 offset 机制。

ACK 机制

转自 javakeeper
为了保证消息的可靠性，消息队列都会有完善的消息确认机制（Acknowledge），即消费者向队列报告消息已收到或已处理的机制。
RPOPLPUSH、BRPOPLPUSH （阻塞）从一个 list 中获取消息的同时把这条消息复制到另一个 list 里（可以当做备份），而且这个过程是原子的。
这样我们就可以在业务流程安全结束后，再删除队列元素，实现消息确认机制。

127.0.0.1:6379> rpush myqueue one
(integer) 1
127.0.0.1:6379> rpush myqueue two
(integer) 2
127.0.0.1:6379> rpush myqueue three
(integer) 3
127.0.0.1:6379> rpoplpush myqueue queuebak
"three"
127.0.0.1:6379> lrange myqueue 0 -1
1) "one"
2) "two"
127.0.0.1:6379> lrange queuebak 0 -1
1) "three"