Redis专题-队列
首先,想一想 Redis 适合做消息队列吗?
1、消息队列的消息存取需求是什么?redis中的解决方案是什么?
无非就是下面这几点:
0、数据可以顺序读取
1、支持阻塞等待拉取消息
2、支持发布/订阅模式
3、重新消费
4、消息不丢失
5、消息可堆积
那我们来看看redis怎么满足这些需求
1.1、基于 List 的消息队列解决方案
1.1.1、数据保证顺序
List 本身就是按先进先出的顺序对数据进行存取的,底层的实现就是一个「链表」,在头部和尾部操作元素,时间复杂度都是 O(1),这意味着它非常符合消息队列的模型。
生产者使用 LPUSH 发布消息:
127.0.0.1:6379> LPUSH mq 5
(integer) 1
127.0.0.1:6379> LPUSH mq 3
(integer) 2
消费者使用 RPOP 拉取消息:
127.0.0.1:6379> RPOP mq
5
127.0.0.1:6379> RPOP mq
3
当队列中已经没有消息了,消费者在执行 RPOP 时,会返回 NULL。
127.0.0.1:6379> RPOP mq
(nil)
消费者读取数据时,有一个潜在的性能风险点:
生产者写入数据时,List 并不会主动通知消费者有新消息写入。
如果消费者想要及时处理消息,需要在程序中不停地调用 RPOP 命令。
如果有新消息写入,RPOP 命令就会返回结果,否则,RPOP 命令返回空值,再继续循环。
// 伪代码
while (true)
{
var msg = redis.rpop("mq")
if(msg == null)
continue;
handle(msg)
}
上述代码中如果队列为空,消费者依旧会频繁拉取消息,这会造成「CPU 空转」,不仅浪费 CPU 资源,还会对 Redis 造成压力。
我们处理一下,当队列为空时,我们可以「休眠」一会,再去尝试拉取消息。
// 伪代码
while (true)
{
var msg = redis.rpop("mq")
if(msg == null)
{
Thread.Sleep(2000);
continue;
}
handle(msg)
}
「CPU 空转」解决了,但是有新的问题发生了:当消费者在休眠等待时有新消息,那么消费者处理新消息就会存在「延迟」。
那如何做,既能及时处理新消息,还能避免 CPU 空转呢?
1.1.2、支持阻塞等待拉取消息
为了解决这个问题,Redis 提供了 BRPOP 命令。BRPOP 命令也称为阻塞式读取,客户端在没有读到队列数据时,自动阻塞,直到有新的数据写入队列,再开始读取新数据。和消费者程序自己不停地调用 RPOP 命令相比,这种方式能节省 CPU 开销。(这里的 B 指的是阻塞(Block)。)
使用 BRPOP 这种阻塞式方式拉取消息时,还支持传入一个「超时时间」,如果设置为 0,则表示不设置超时,直到有新消息才返回,否则会在指定的超时时间后返回 NULL
// 伪代码
while (true)
{
// 没消息阻塞等待,0表示不设置超时时间
var msg = redis.brpop("mq",0)
if(msg == null)
continue;
handle(msg)
}
注意:如果设置的超时时间太长,这个连接太久没有活跃过,可能会被 Redis Server 判定为无效连接,之后 Redis Server 会强制把这个客户端踢下线。所以,采用这种方案,客户端要有重连机制。
1.1.3、发布/订阅模式
不支持。
1.1.4、重新消费
不支持。
但是在业务使用唯一ID等方式实现,消费ID后做判断是否处理过,使对于同一条消息处理结果都是一致的,保证幂等性。
1.1.5、消息不丢失
仅消费端不丢失。
List 类型提供了 BRPOPLPUSH 命令,这个命令的作用是让消费者程序从一个 List 中读取消息,同时,Redis 会把这个消息再插入到另一个 List(可以叫作备份 List)留存。
如果消费者程序读了消息但没能正常处理,等它重启后,就可以从备份 List 中重新读取消息并进行处理了。
1.1.6、消息堆积
不可堆积。
如果消费较慢,List 中的消息越积越多,redis内存压力会越来越大。
而且List本身也不支持消费组,不能使用多个消费端消费。
1.1.7、小结
需求 | LIST |
---|---|
数据保证顺序 | 支持。使用LPUSH/RPOP |
支持阻塞等待拉取消息 | 支持。使用BRPOP |
支持发布 / 订阅模式 | 不支持 |
重复消费 | 不支持。但是可以自行实现全局唯一ID |
消息不丢失 | 不完全。消费端算是不丢失,BRPOPLPUSH |
消息堆积 | 不支持。内存持续增长 |
简单的业务场景,可以使用list。
但如果想要有多个生产者和消费者,那么可以继续往下看。
1.2、基于 Pub/Sub 的消息队列解决方案
Redis 专门是针对「发布/订阅」( PUBLISH / SUBSCRIBE) 这种队列模型设计的。
可以解决重复消费问题,可以多组生产者、消费者场景。
使用 Pub/Sub 这种方案,既支持阻塞式拉取消息,还很好地满足了多组消费者,消费同一批数据的业务需求。
除此之外,Pub/Sub 还提供了「匹配订阅」模式,允许消费者根据一定规则,订阅「多个」自己希望的队列。
可以看到,Pub/Sub 最大的优势就是,支持多组生产者、消费者处理消息。
缺点就是:丢数据。
Pub/Sub 没有基于任何数据类型,也没有做任何的数据存储(不会写入到 RDB 和 AOF 中),单纯的建立转发通道,把符合规则的数据转发到另外一端,一切都是实时转发的。
如果消费者异常,那么再次上线只能接受新的消息,在此期间生产者找不到消费者就会丢弃数据。
使用 Pub/Sub 时,注意:消费者必须先订阅队列,生产者才能发布消息,否则消息会丢失。
消息积压时消息也可能会消息丢失或者消费失败,Pub/Sub的实现上就是在server的内存上给订阅的消费者分配了一个buffer。
生产者发布消息不断写入buffer中,当消息积压时,buffer占用内存会持续增长,如果突破了buffer配置的上线,那么消费者就会被踢下线,导致消费失败,数据丢失。
缓冲区的默认配置:client-output-buffer-limit pubsub 32mb 8mb 60。
32mb:缓冲区一旦超过 32MB,Redis 直接强制把消费者踢下线.
8mb + 60:缓冲区超过 8MB,并且持续 60 秒,Redis 也会把消费者踢下线
List 拉数据,Pub/Sub推数据。
Pub/Sub 的优缺点:
1、支持发布 / 订阅,支持多组生产者、消费者处理消息
2、消费者下线,数据会丢失
3、不支持数据持久化,Redis 宕机,数据也会丢失
4、消息堆积,缓冲区溢出,消费者会被强制踢下线,数据也会丢失
哨兵集群和 Redis 实例通信时,采用了 Pub/Sub 的方案,因为哨兵正好符合即时通讯的业务场景。
很明显Pub/Sub不是我们想要的消息队列,继续往下看
1.3、基于 Streams 的消息队列解决方案
Streams 是 Redis 专门为消息队列设计的数据类型,它提供了丰富的消息队列操作命令。
XADD:插入消息,保证有序,可以自动生成全局唯一ID
XREAD:用于读取消息,可以按ID读取数据
XREADGROUP:按消费组形式读取消息
XPENDING:可以用来查询每个消费组内所有消费者已读取但尚未确认的消息
XACK:用于向消息队列确认消息处理已完成
生产者推消息:
// *表示让Redis自动生成消息ID
127.0.0.1:6379> XADD queue * name zhangsan
"1618469123380-0"
127.0.0.1:6379> XADD queue * name lisi
"1618469127777-0"
消费者拉消息:
XADD「*」表示让 Redis 自动生成唯一的消息 ID
消息 ID 的格式是「时间戳-自增序号」(自增序号从0开始编号)
// 从开头读取5条消息,0-0表示从开头读取
127.0.0.1:6379> XREAD COUNT 5 STREAMS queue 0-0
1) 1) "queue"
2) 1) 1) "1618469123380-0"
2) 1) "name"
2) "zhangsan"
2) 1) "1618469127777-0"
2) 1) "name"
2) "lisi"
如果想继续拉取消息,需要传入上一条消息的 ID:
127.0.0.1:6379> XREAD COUNT 5 STREAMS queue 1618469127777-0
(nil)
这就是Stream 最简单的生产、消费。
1.3.1、数据保证顺序
支持。
XADD插入消息,保证有序
1.3.2、支持阻塞等待拉取消息
支持。
在读取消息时,只需要增加 BLOCK 参数即可。
// BLOCK 0 表示阻塞等待,不设置超时时间
127.0.0.1:6379> XREAD COUNT 5 BLOCK 0 STREAMS queue 1618469127777-0
这时,消费者就会阻塞等待,直到生产者发布新的消息才会返回。
1.3.3、发布/订阅模式
支持。
Stream 通过以下命令完成发布订阅:
XGROUP:创建消费者组
XREADGROUP:在指定消费组下,开启消费者拉取消息
127.0.0.1:6379> XADD queue * name zhangsan
"1618470740565-0"
127.0.0.1:6379> XADD queue * name lisi
"1618470743793-0"
// 创建消费者组1,0-0表示从头拉取消息
127.0.0.1:6379> XGROUP CREATE queue group1 0-0
OK
// 创建消费者组2,0-0表示从头拉取消息
127.0.0.1:6379> XGROUP CREATE queue group2 0-0
OK
第一个消费组开始消费:
// group1的consumer开始消费,>表示拉取最新数据
127.0.0.1:6379> XREADGROUP GROUP group1 consumer COUNT 5 STREAMS queue >
1) 1) "queue"
2) 1) 1) "1618470740565-0"
2) 1) "name"
2) "zhangsan"
2) 1) "1618470743793-0"
2) 1) "name"
2) "lisi"
同样地,第二个消费组开始消费:
// group2的consumer开始消费,>表示拉取最新数据
127.0.0.1:6379> XREADGROUP GROUP group2 consumer COUNT 5 STREAMS queue >
1) 1) "queue"
2) 1) 1) "1618470740565-0"
2) 1) "name"
2) "zhangsan"
2) 1) "1618470743793-0"
2) 1) "name"
2) "lisi"
我们可以看到,这 2 组消费者,都可以获取同一批数据进行处理了。
通过创建消费组的形式达到订阅的目的。
1.3.4、重新消费
支持。
上面拉取消息时用到了消息 ID,这里为了保证重新消费,也要用到这个消息 ID。
当一组消费者处理完消息后,需要执行 XACK 命令告知 Redis,这时 Redis 就会把这条消息标记为「处理完成」。
// group1下的 1618472043089-0 消息已处理完成
127.0.0.1:6379> XACK queue group1 1618472043089-0
如果消费者异常宕机,肯定不会发送 XACK,那么 Redis 就会依旧保留这条消息。
待这组消费者重新上线后,Redis 就会把之前没有处理成功的数据,重新发给这个消费者。这样一来,即使消费者异常,也不会丢失数据了。
// 消费者重新上线,0-0表示重新拉取未ACK的消息
127.0.0.1:6379> XREADGROUP GROUP group1 consumer1 COUNT 5 STREAMS queue 0-0
// 之前没消费成功的数据,依旧可以重新消费
1) 1) "queue"
2) 1) 1) "1618472043089-0"
2) 1) "name"
2) "zhangsan"
2) 1) "1618472045158-0"
2) 1) "name"
2) "lisi"
1.3.5、消息不丢失
Stream 是新增加的数据类型,它与其它数据类型一样,每个写操作,也都会写入到 RDB 和 AOF 中。
我们只需要配置好持久化策略,这样的话,就算 Redis 宕机重启,Stream 中的数据也可以从 RDB 或 AOF 中恢复回来。
1.3.6、消息堆积
支持,但有长度限制。
当消息队列发生消息堆积时,一般只有 2 个解决方案:
1、生产者限流:避免消费者处理不及时,导致持续积压
2、丢弃消息:中间件丢弃旧消息,只保留固定长度的新消息
Redis 在实现 Stream 时,采用了第 2 个方案。
在发布消息时,你可以指定队列的最大长度,防止队列积压导致内存爆炸。
// 队列长度最大10000
127.0.0.1:6379> XADD queue MAXLEN 10000 * name zhangsan
"1618473015018-0"
当队列长度超过上限后,旧消息会被删除,只保留固定长度的新消息。
这么来看,Stream 在消息积压时,如果指定了最大长度,还是有可能丢失消息的。
除了以上介绍到的命令,Stream 还支持查看消息长度(XLEN)、查看消费者状态(XINFO)等命令
1.3.7、小结
需求 | Stream |
---|---|
数据保证顺序 | 支持 |
支持阻塞等待拉取消息 | 支持 |
支持发布 / 订阅模式 | 支持 |
重复消费 | 支持 |
消息不丢失 | 支持 |
消息堆积 | 支持 |
既然它的功能这么强大,这是不是意味着,Redis 真的可以作为专业的消息队列中间件来使用呢?
2、与专业的消息队列对比
一个专业的消息队列,必须要做到两大块:
1、消息不丢
2、消息可堆积
消息队列,其实就分为三大块:生产者、队列中间件、消费者。
2.1、如何保证不丢消息?
2.1.1、生产者会不会丢失数据?
生产者丢失:
1、消息没法出去,网络原因或者其他原因,中间件返回失败
2、不确定是否发送成功:网络原因等导致发布超时,数据可能已经发送成功,但读取响
应超时
第一种情况,重发即可。
第二种情况,因为不知道是否成功,为了避免丢失,就只能也重试发送到成功为止。
生产者一般设定重试次数,超过上限次数需记录日志,发送警报。
是的,为了不丢失,可以接受重复发送,在消费端就需要做一些逻辑判断了,业务可能需要保证幂等性。
所以,redis或者其他中间件队列,都可以在生产者上保证不丢失数据。
2.1.2、消费者会不会丢失数据?
消费者拿到消息后,还没处理完成,就异常宕机了,那消费者还能否重新消费失败的消息?
要解决这个问题,消费者在处理完消息后,必须「告知」队列中间件,队列中间件才会把标记已处理,否则仍旧把这些数据发给消费者。
这种方案需要消费者和中间件互相配合,才能保证消费者这一侧的消息不丢。
无论是 Redis 的 Stream,还是专业的队列中间件,例如 RabbitMQ、Kafka,其实都是这么做的。
所以,从这个角度来看,Redis 也是合格的。
2.1.3、队列中间件会不会丢失数据?
上面的问题只要客户端和服务端配合好,就能保证生产端、消费端都不丢消息。
但是,如果队列中间件本身就不可靠呢?
在这个方面,Redis 其实没有达到要求。
Redis 在以下 2 个场景下,都会导致数据丢失。
1、AOF 持久化配置为每秒写盘,但这个写盘过程是异步的,Redis 宕机时会存在数据丢失的可能
2、主从复制也是异步的,主从切换时,也存在丢失数据的可能(从库还未同步完成主库发来的数据,就被提成主库)
基于以上原因我们可以看到,Redis 本身的无法保证严格的数据完整性。
RabbitMQ 或 Kafka 这类专业的队列中间件,在使用时,一般是部署一个集群,生产者在发布消息时,队列中间件通常会写「多个节点」,以此保证消息的完整性。这样一来,即便其中一个节点挂了,也能保证集群的数据不丢失。
Redis 的定位则不同,它的定位更多是当作缓存来用,它们两者在这个方面肯定是存在差异的。
2.1.4、消息积压怎么办?
Redis 的数据都存储在内存中,这就意味着一旦发生消息积压,则会导致 Redis 的内存持续增长,如果超过机器内存上限,就会面临被 OOM 的风险。
Redis 的 Stream 提供了可以指定队列最大长度的功能,就是为了避免这种情况发生。
但 Kafka、RabbitMQ 这类消息队列就不一样了,它们的数据都会存储在磁盘上,磁盘的成本要比内存小得多,当消息积压时,无非就是多占用一些磁盘空间,相比于内存,在面对积压时也会更加「坦然」。
把 Redis 当作队列来使用时,始终面临的 2 个问题:
1、Redis 本身可能会丢数据,
2、面对消息积压 Redis 内存资源紧张.
如果你的业务场景足够简单,对于数据丢失不敏感,而且消息积压概率比较小的情况下,把 Redis 当作队列是完全可以的。
而且,Redis 相比于 Kafka、RabbitMQ,部署和运维也更加轻量。
如果你的业务场景对于数据丢失非常敏感,而且写入量非常大,消息积压时会占用很多的机器资源,那么我建议你使用专业的消息队列中间件。
3、额外补充
3.1、延迟队列
应用场景:
1、订单超时未支付,关闭订单退还库存
2、订单完成5天后没有评论自动好评
3、用户并发量大,延后发送邮件短信
4、.....
3.1.1实现方式
-
ZSET + 定时轮询
- zset支持高性能的 score 排序,且去重
- 内存上进行操作的,速度非常快
- 注意多进程争抢,使用lua将zrangebyscore和zrem进行原子化
-
监听key(不建议)
- WATCH 可以鉴定单个或者多个key的变化情况
- 数量较大时,监听会滞后(过期事件是在Redis服务器删除密钥时产生的,而不是在理论上存活时间达到零时产生的)
参考、复制、学习、引用与:
标签:专题,消费者,队列,Redis,6379,丢失,消息 From: https://www.cnblogs.com/hanfan/p/17631061.htmlredis官网
请勿过度依赖 Redis 的过期监听
把Redis当作队列来用,真的合适吗?
消息队列的考验:Redis有哪些解决方案?