关于「把 Redis 当作队列来用是否合适」的问题。
有些人表示赞成,他们认为 Redis 很轻量,用作队列很方便。
也些人则反对,认为 Redis 会「丢」数据,最好还是用「专业」的队列中间件更稳妥。
从最简单的开始:List 队列
首先,我们先从最简单的场景开始讲起。
如果你的业务需求足够简单,想把 Redis 当作队列来使用,肯定最先想到的就是使用 List 这个数据类型。
因为 List 底层的实现就是一个「链表」,在头部和尾部操作元素,时间复杂度都是 O(1),这意味着它非常符合消息队列的模型。
如果把 List 当作队列,你可以这么来用。
生产者使用 LPUSH 发布消息:
127.0.0.1:6379> LPUSH queue msg1
(integer) 1
127.0.0.1:6379> LPUSH queue msg2
(integer) 2
消费者这一侧,使用 RPOP 拉取消息:
127.0.0.1:6379> RPOP queue
"msg1"
127.0.0.1:6379> RPOP queue
"msg2"
这个模型非常简单,也很容易理解。
但这里有个小问题,当队列中已经没有消息了,消费者在执行 RPOP 时,会返回 NULL。
127.0.0.1:6379> RPOP queue
(nil) // 没消息了
而我们在编写消费者逻辑时,一般是一个「死循环」,这个逻辑需要不断地从队列中拉取消息进行处理,伪代码一般会这么写:
while true:
msg = redis.rpop("queue")
// 没有消息,继续循环
if msg == null:
continue
// 处理消息
handle(msg)
如果此时队列为空,那消费者依旧会频繁拉取消息,这会造成「CPU 空转」,不仅浪费 CPU 资源,还会对 Redis 造成压力。
怎么解决这个问题呢?
也很简单,当队列为空时,我们可以「休眠」一会,再去尝试拉取消息。代码可以修改成这样:
while true:
msg = redis.rpop("queue")
// 没有消息,休眠2s
if msg == null:
sleep(2)
continue
// 处理消息
handle(msg)
这就解决了 CPU 空转问题。
这个问题虽然解决了,但又带来另外一个问题:当消费者在休眠等待时,有新消息来了,那消费者处理新消息就会存在「延迟」。
假设设置的休眠时间是 2s,那新消息最多存在 2s 的延迟。
要想缩短这个延迟,只能减小休眠的时间。但休眠时间越小,又有可能引发 CPU 空转问题。
鱼和熊掌不可兼得。
那如何做,既能及时处理新消息,还能避免 CPU 空转呢?
Redis 是否存在这样一种机制:如果队列为空,消费者在拉取消息时就「阻塞等待」,一旦有新消息过来,就通知我的消费者立即处理新消息呢?
幸运的是,Redis 确实提供了「阻塞式」拉取消息的命令:BRPOP / BLPOP,这里的 B 指的是阻塞(Block)。
现在,你可以这样来拉取消息了:
while true:
// 没消息阻塞等待,0表示不设置超时时间
msg = redis.brpop("queue", 0)
if msg == null:
continue
// 处理消息
handle(msg)
使用 BRPOP 这种阻塞式方式拉取消息时,还支持传入一个「超时时间」,如果设置为 0,则表示不设置超时,直到有新消息才返回,否则会在指定的超时时间后返回 NULL。
这个方案不错,既兼顾了效率,还避免了 CPU 空转问题,一举两得。
注意:如果设置的超时时间太长,这个连接太久没有活跃过,可能会被 Redis Server 判定为无效连接,之后 Redis Server 会强制把这个客户端踢下线。所以,采用这种方案,客户端要有重连机制。
解决了消息处理不及时的问题,你可以再思考一下,这种队列模型,有什么缺点?
我们一起来分析一下:
-
不支持重复消费:消费者拉取消息后,这条消息就从 List 中删除了,无法被其它消费者再次消费,即不支持多个消费者消费同一批数据
-
消息丢失:消费者拉取到消息后,如果发生异常宕机,那这条消息就丢失了
第一个问题是功能上的,使用 List 做消息队列,它仅仅支持最简单的,一组生产者对应一组消费者,不能满足多组生产者和消费者的业务场景。
第二个问题就比较棘手了,因为从 List 中 POP 一条消息出来后,这条消息就会立即从链表中删除了。也就是说,无论消费者是否处理成功,这条消息都没办法再次消费了。
这也意味着,如果消费者在处理消息时异常宕机,那这条消息就相当于丢失了。
下面是基于php的一个简单的redis阻塞队列的demo
生产者:
// redis写入数据
redis()->lpush('ceshi', "ceshi");
消费者:
$redisKey = "ceshi";
$timeout = 20; // 阻塞时间: s
while (true) {
try {
$res = redis()->brpop($redisKey, $timeout);
if (!empty($res)) {
// 业务逻辑
}
} catch (\Exception $e) {
// 异常处理
// 如果redis连接超时,会抛出异常,可以在这里进行重连
}
sleep(1);
}
今天的文章Redis队列_redis面试必会6题经典[通俗易懂]分享到此就结束了,感谢您的阅读。
版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。
如需转载请保留出处:https://bianchenghao.cn/77237.html