1 Redis 延时队列,一次性搞明白-德赢Vwin官网 网
0
  • 聊天消息
  • 系统消息
  • 评论与回复
登录后你可以
  • 下载海量资料
  • 学习在线课程
  • 观看技术视频
  • 写文章/发帖/加入社区
会员中心
创作中心

完善资料让更多小伙伴认识你,还能领取20积分哦,立即完善>

3天内不再提示

Redis 延时队列,一次性搞明白

数据分析与开发 来源:数据分析与开发 作者:数据分析与开发 2020-10-30 16:34 次阅读

所谓延时队列就是延时的消息队列,下面说一下一些业务场景

实践场景

订单支付失败,每隔一段时间提醒用户

用户并发量的情况,可以延时2分钟给用户发短信

先来看看Redis实现普通的消息队列

我们知道,对于专业的消息队列中间件,如Kafka和RabbitMQ,消费者在消费消息之前要进行一系列的繁琐过程。

如RabbitMQ发消息之前要创建 Exchange,再创建 Queue,还要将 Queue 和 Exchange 通过某种规则绑定起来,发消息的时候要指定 routingkey,还要控制头部信息

但是绝大 多数情况下,虽然我们的消息队列只有一组消费者,但还是需要经历上面一些过程。

有了 Redis,对于那些只有一组消费者的消息队列,使用 Redis 就可以非常轻松的搞定。Redis 的消息队列不是专业的消息队列,它没有非常多的高级特性, 没有 ack 保证,如果对消息的可靠性有着极致的追求,那么它就不适合使用

异步消息队列基本实现

Redis 的 list(列表) 数据结构常用来作为异步消息队列使用,使用 rpush/lpush 操作入队列, 使用 lpop 和 rpop 来出队列

>rpushqueue月伴飞鱼1月伴飞鱼2月伴飞鱼3 (integer)3 >lpopqueue "月伴飞鱼1" >llenqueue (integer)2

问题1:如果队列空了

客户端是通过队列的 pop 操作来获取消息,然后进行处理。处理完了再接着获取消息, 再进行处理。如此循环往复,这便是作为队列消费者的客户端的生命周期。

可是如果队列空了,客户端就会陷入 pop 的死循环,不停地 pop,没有数据,接着再 pop, 又没有数据。这就是浪费生命的空轮询。空轮询不但拉高了客户端的 CPU,redis 的 QPS 也 会被拉高,如果这样空轮询的客户端有几十来个,Redis 的慢查询可能会显著增多。

通常我们使用 sleep 来解决这个问题,让线程睡一会,睡个 1s 钟就可以了。不但客户端 的 CPU 能降下来,Redis 的 QPS 也降下来了

问题2:队列延迟

用上面睡眠的办法可以解决问题。同时如果只有 1 个消费者,那么这个延迟就是 1s。如果有多个消费者,这个延迟会有所下降,因 为每个消费者的睡觉时间是岔开来的。

有没有什么办法能显著降低延迟呢?

那就是 blpop/brpop。

这两个指令的前缀字符 b 代表的是 blocking,也就是阻塞读。

阻塞读在队列没有数据的时候,会立即进入休眠状态,一旦数据到来,则立刻醒过来。消 息的延迟几乎为零。用 blpop/brpop 替代前面的 lpop/rpop,就完美解决了上面的问题。

问题3:空闲连接自动断开

其实他还有个问题需要解决—— 空闲连接的问题。

如果线程一直阻塞在哪里,Redis 的客户端连接就成了闲置连接,闲置过久,服务器一般 会主动断开连接,减少闲置资源占用。这个时候 blpop/brpop 会抛出异常来。

所以编写客户端消费者的时候要小心,注意捕获异常,还要重试。

分布式锁冲突处理

假如客户端在处理请求时加分布式锁没加成功怎么办。

一般有 3 种策略来处理加锁失败:

1、直接抛出异常,通知用户稍后重试;

2、sleep 一会再重试;

3、将请求转移至延时队列,过一会再试;

直接抛出特定类型的异常

这种方式比较适合由用户直接发起的请求,用户看到错误对话框后,会先阅读对话框的内 容,再点击重试,这样就可以起到人工延时的效果。如果考虑到用户体验,可以由前端的代码 替代用户自己来进行延时重试控制。它本质上是对当前请求的放弃,由用户决定是否重新发起 新的请求。

sleep

sleep 会阻塞当前的消息处理线程,会导致队列的后续消息处理出现延迟。如果碰撞的比 较频繁或者队列里消息比较多,sleep 可能并不合适。如果因为个别死锁的 key 导致加锁不成 功,线程会彻底堵死,导致后续消息永远得不到及时处理。

延时队列

这种方式比较适合异步消息处理,将当前冲突的请求扔到另一个队列延后处理以避开冲突。

延时队列的实现

我们可以使用 zset这个命令,用设置好的时间戳作为score进行排序,使用zadd score1 value1 ....命令就可以一直往内存中生产消息。再利用 zrangebysocre 查询符合条件的所有待处理的任务,通过循环执行队列任务即可。也可以通过zrangebyscore key min max withscores limit 0 1查询最早的一条任务,来进行消费

privateJedisjedis; publicvoidredisDelayQueueTest(){ Stringkey="delay_queue"; //实际开发建议使用业务ID和随机生成的唯一ID作为value,随机生成的唯一ID可以保证消息的唯一性,业务ID可以避免value携带的信息过多 StringorderId1=UUID.randomUUID().toString(); jedis.zadd(queueKey,System.currentTimeMillis()+5000,orderId1); StringorderId12=UUID.randomUUID().toString(); jedis.zadd(queueKey,System.currentTimeMillis()+5000,orderId2); newThread(){ @Override publicvoidrun(){ while(true){ SetresultList; //只获取第一条数据,只获取不会移除数据 resultList=jedis.zrangebyscore(key,System.currentTimeMillis(),0,1); if(resultList.size()==0){ try{ Thread.sleep(1000); }catch(InterruptedExceptione){ e.printStackTrace(); break; } }else{ //移除数据获取到的数据 if(jedis.zrem(key,resultList.get(0))>0){ StringorderId=resultList.get(0); log.info("orderId={}",resultList.get(0)); this.handleMsg(orderId); } } } } }.start(); } publicvoidhandleMsg(Tmsg){ System.out.println(msg); }

上面的实现, 在多线程逻辑上也是没有问题的, 假设有两个线程 T1, T2和其他更多线程, 处理逻辑如下, 保证了多线程情况下只有一个线程处理了对应的消息:

1.T1, T2 和其他更多线程调用 zrangebyscore 获取到了一条消息 A

2.T1 准备开始删除消息 A, 由于是原子操作, T2 和其他更多线程等待 T1 执行 zrem 删除消息 A 后再执行 zrem 删除消息 A

3.T1 删除了消息 A, 返回删除成功标记 1, 并对消息 A 进行处理

4.T2 其他更多线程开始 zrem 删除消息 A, 由于消息 A 已经被删除, 所以所有的删除均失败, 放弃了对消息 A 的处理

同时,我们要注意一定要对handle_msg进行异常捕获,避免因为个别任务处理问题导致循环异常退 出

进一步优化

上面的算法中同一个任务可能会被多个进程取到之后再使用 zrem 进行争抢,那些没抢到 的进程都是白取了一次任务,这是浪费。可以考虑使用 lua scripting 来优化一下这个逻辑,将 zrangebyscore 和 zrem 一同挪到服务器端进行原子化操作,这样多个进程之间争抢任务时就不 会出现这种浪费了

使用调用Lua脚本进一步优化

Lua 脚本, 如果有超时的消息, 就删除, 并返回这条消息, 否则返回空字符串:

StringluaScript="localresultArray=redis.call('zrangebyscore',KEYS[1],0,ARGV[1],'limit',0,1) "+ "if#resultArray>0then "+ "ifredis.call('zrem',KEYS[1],resultArray[1])>0then "+ "returnresultArray[1] "+ "else "+ "return'' "+ "end "+ "else "+ "return'' "+ "end"; jedis.eval(luaScript,ScriptOutputType.VALUE,newString[]{key},String.valueOf(System.currentTimeMillis()));

Redis延时队列优势

Redis用来进行实现延时队列是具有这些优势的:

1.Redis zset支持高性能的 score 排序。

2.Redis是在内存上进行操作的,速度非常快。

3.Redis可以搭建集群,当消息很多时候,我们可以用集群来提高消息处理的速度,提高可用性。

4.Redis具有持久化机制,当出现故障的时候,可以通过AOF和RDB方式来对数据进行恢复,保证了数据的可靠性

Redis延时队列劣势

使用 Redis 实现的延时消息队列也存在数据持久化, 消息可靠性的问题

没有重试机制 - 处理消息出现异常没有重试机制, 这些需要自己去实现, 包括重试次数的实现等

没有 ACK 机制 - 例如在获取消息并已经删除了消息情况下, 正在处理消息的时候客户端崩溃了, 这条正在处理的这些消息就会丢失, MQ 是需要明确的返回一个值给 MQ 才会认为这个消息是被正确的消费了

如果对消息可靠性要求较高, 推荐使用 MQ 来实现

Redission实现延时队列

基于Redis的Redisson分布式延迟队列结构的RDelayedQueue Java对象在实现了RQueue接口的基础上提供了向队列按要求延迟添加项目的功能。该功能可以用来实现消息传送延迟按几何增长或几何衰减的发送策略

RQueuedistinationQueue=... RDelayedQueuedelayedQueue=getDelayedQueue(distinationQueue); //10秒钟以后将消息发送到指定队列 delayedQueue.offer("msg1",10,TimeUnit.SECONDS); //一分钟以后将消息发送到指定队列 delayedQueue.offer("msg2",1,TimeUnit.MINUTES);

在该对象不再需要的情况下,应该主动销毁。仅在相关的Redisson对象也需要关闭的时候可以不用主动销毁。

RDelayedQueuedelayedQueue=... delayedQueue.destroy();

是不是很方便...............

责任编辑:xj

原文标题:Redis 延时队列,这次彻底给你整明白了

文章出处:【微信公众号:数据分析与开发】欢迎添加关注!文章转载请注明出处。

声明:本文内容及配图由入驻作者撰写或者入驻合作网站授权转载。文章观点仅代表作者本人,不代表德赢Vwin官网 网立场。文章及其配图仅供工程师学习之用,如有内容侵权或者其他违规问题,请联系本站处理。 举报投诉
  • 编程
    +关注

    关注

    88

    文章

    3614

    浏览量

    93685
  • 线程
    +关注

    关注

    0

    文章

    504

    浏览量

    19675
  • Redis
    +关注

    关注

    0

    文章

    374

    浏览量

    10871

原文标题:Redis 延时队列,这次彻底给你整明白了

文章出处:【微信号:DBDevs,微信公众号:数据分析与开发】欢迎添加关注!文章转载请注明出处。

收藏 人收藏

    评论

    相关推荐

    ADS1256第一次上电的时候,采集的ADC信号是实际值的半,为什么?

    会几个小时复现一次,有时候会几天复现一次,真的明白哪里出问题,希望各位大神帮忙解决下。具体会在哪些地方出问题。
    发表于 12-13 15:33

    智芯传感ZXP4系列一次性血压传感器介绍

    智芯传感ZXP4系列一次性血压传感器,主要供医疗单位对患者进行动脉压、中心静脉压、肺动脉压、左冠状动脉压等多种血压监测,属于有创血压监测。
    的头像 发表于 12-04 13:52 618次阅读
    智芯传感ZXP4系列<b class='flag-5'>一次性</b>血压传感器介绍

    【AI技术支持】USB_CDC电脑串口一次性发送100000byte丢包问题处理

    启明云端/01你是否曾遇到过?在使用ESP32-S3,ESP-IDF版本为idf5.2时,蒸汽锅产品基于例程tusb_serial_device测试USBCDC自发自收,电脑CDC串口一次性发送
    的头像 发表于 11-07 08:01 265次阅读
    【AI技术支持】USB_CDC电脑串口<b class='flag-5'>一次性</b>发送100000byte丢包问题处理

    一次电源与二电源有什么不同

    在电力系统和电子设备的供电领域中,一次电源与二电源是两个至关重要的概念。它们各自承担着不同的功能和角色,共同确保电力供应的稳定性和可靠。本文将对一次电源与二
    的头像 发表于 10-10 14:10 1724次阅读

    ODU MEDI-SNAP一次性医用插拔自锁插头产品介绍

    为满足一次性内窥镜、一次性手术消融刀等设备中的耗材需求,欧度全新推出了MEDI-SNAP一次性医用插拔自锁插头,为医疗客户打造了组在品质与经济
    的头像 发表于 09-10 09:59 431次阅读

    aP8942A一次性编程(OTP)语音集成电路英文手册

    AP8942A是一次性编程(OTP)语音集成电路(IC),它设计用于在各种应用中存储和播放语音消息。以下是AP8942A语音IC的详细介绍:核心特性存储容量:内嵌1M bits的EPROM,可以
    发表于 08-08 14:40 4次下载

    AP8921A一次性编程(OTP)语音集成电路英文手册

    AP8921A是一次性编程(OTP)语音集成电路(IC),它具备多种功能,适用于多种需要语音提示或声音播放的应用场景。以下是AP8921A语音IC的详细介绍:核心特性存储容量:内嵌512k
    发表于 08-08 14:39 0次下载

    esp32如何一次性读取大文件数据?

    esp32没有提供数据库读写的例子,最近有个大文件,无法一次性读出,请问,怎么读取,json中部分json数组。然后修改完了以后,在写入进去?
    发表于 06-25 06:52

    一次性注射针刚性测试仪作用与重要

    文章由济南三泉智能科技有限公司提供一次性注射针刚性测试仪是用于评估一次性注射针刚性性能的专用设备。、作用与重要确保注射针质量:注射针的刚性是评价其质量的重要参数之
    的头像 发表于 06-04 15:32 322次阅读
    <b class='flag-5'>一次性</b>注射针刚性测试仪作用与重要<b class='flag-5'>性</b>

    请问如何将TIM1_CCR1H和TIM1_CCR1L合并成个变量,之后一次性赋值?

    请问:TIM1_CCR1H和TIM1_CCR1L分别是比较寄存器的高八位和第八位,用来设置PWM的占空比的。是否可以把这两个寄存器变成个整体,之后一次性对其进行赋值?
    发表于 05-10 07:47

    一次性使用无菌脑积水分流器综合测试仪

    一次性使用无菌脑积水分流器综合测试仪 描述 一次性使用无菌脑积水分流器是种用于治疗脑积水的医疗器械。脑积水是指在颅内或脑脊液循环系统中积聚过多液体导致颅内压力增高的病症。脑积水分流
    的头像 发表于 04-06 09:13 331次阅读
    <b class='flag-5'>一次性</b>使用无菌脑积水分流器综合<b class='flag-5'>性</b>测试仪

    九齐语音芯片NY5P025B SOP8 SOP16 OTP一次性烧录IC方案开发

    九齐语音芯片NY5P025B SOP8 SOP16 OTP一次性烧录IC方案开发 、引言 随着电子科技的不断发展,语音芯片在各种电子产品中的应用越来越广泛。九齐科技的NY5P025B
    的头像 发表于 03-13 14:39 946次阅读

    GB 15811一次性医用注射针检测仪器

    GB 15811一次性医用注射针检测仪器
    发表于 02-18 14:21 0次下载

    nuc980从机一次性需要把许多数据发给上位机,每发个packet必须长时间延时,是什么原因导致的?

    , .unbind = hid_unbind, }; 把高速改成全速,但是好像没起到作用,设备信息里还是高速。请问,从机一次性需要把许多数据发给上位机,每发个packet必须长时间延时,这是什么原因导致的。
    发表于 01-17 06:26

    一次性保险丝与自恢复保险丝这两种之间有什么差别?

    一次性保险丝与自恢复保险丝这两种之间有什么差别? 一次性保险丝和自恢复保险丝是两种用于电气系统中的保险器件,它们在功能和特性上存在些差异。本文将对这两种保险丝进行比较,并详细解释它们的定义
    的头像 发表于 01-04 14:13 1668次阅读