一、应用场景:
- 订单超过 30 分钟未支付,则自动取消。
- 外卖商家超时未接单,则自动取消。
- 医生抢单电话点诊,超过 30 分钟未打电话,则自动退款。
等等场景都可以用定时任务去轮询实现,但是当数据量过大的时候,高频轮询数据库会消耗大量的资源,此时用延迟队列来应对这类场景比较好。
二、需求
- 消息存储
- 过期延时消息实时获取
- 高可用性
三、为什么使用 Redis 实现?
3.1、Rabbitmq 延时队列
- 优点:消息持久化,分布式
- 缺点:延时相同的消息必须扔在同一个队列,每一种延时就需要建立一个队列。因为当后面的消息比前面的消息先过期,还是只能等待前面的消息过期,这里的过期检测是惰性的。
- 使用: RabbitMQ 可以针对 Queue 设置 x-expires 或者针对 Message 设置 x-message-ttl ,来控制消息的生存时间(可以根据 Queue 来设置,也可以根据 message 设置), Queue 还可以配置 x-dead-letter-exchange 和 x-dead-letter-routing-key(可选)两个参数,如果队列内出现了 dead letter ,则按照这两个参数重新路由转发到指定的队列,此时就可以实现延时队列了。
3.2、DelayQueue 延时队列
- 优点:无界、延迟、阻塞队列
- 缺点:非持久化
- 介绍:JDK 自带的延时队列,没有过期元素的话,使用 poll() 方法会返回 null 值,超时判定是通过getDelay(TimeUnit.NANOSECONDS) 方法的返回值小于等于0来判断,并且不能存放空元素。
- 使用:getDelay 方法定义了剩余到期时间,compareTo 方法定义了元素排序规则。poll() 是非阻塞的获取数据,take() 是阻塞形式获取数据。实现 Delayed 接口即可使用延时队列。
- 注意:DelayQueue 实现了 Iterator 接口,但 iterator() 遍历顺序不保证是元素的实际存放顺序。
/**
* 实现 Delayed 定义延时队列
*/
@Data
@NoArgsConstructor
@AllArgsConstructor
public class Sequence implements Delayed {
private Long time;
private String name;
@Override
public long getDelay(TimeUnit unit) {
return time - System.currentTimeMillis();
}
@Override
public int compareTo(Delayed o) {
if (this.getDelay(TimeUnit.MILLISECONDS) > o.getDelay(TimeUnit.MILLISECONDS)) {
return 1;
} else if (this.getDelay(TimeUnit.MILLISECONDS) < o.getDelay(TimeUnit.MILLISECONDS)) {
return -1;
} else {
return 0;
}
}
}
3.3、Scala 的 Await & Future
- 优点:消息实时性
- 缺点:非持久化
- 介绍:Scala 的 ExecutionContext 中使用
Await 的 result(awaitable: Awaitable[T], atMost: Duration)
方法可以根据传入的 atMost 间隔时间异步执行 awaitable。
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.{Await, Future}
object test extends App {
val task = Future{ doSomething() }
Await.result(task, 5 seconds)
}
3.4、Redis 延迟队列
- 消息持久化,消息至少被消费一次
- 实时性:存在一定的时间误差(定时任务间隔)
- 支持指定消息 remove
- 高可用性
- Redis 的特殊数据结构 ZSet 满足延迟的特性
四、Redis 的使用
4.1、使用 sortedset 操作元素
- 赋值:zadd key score1 value1 score2 value2... (把全部的元素添加到sorted set中,并且每个元素有其对应的分数,返回值是新增的元素个数。)
- 获取元素:
- zscore key value:返回指定成员的分数
- zcard key : 获取集合中的成员数量
- 删除元素:zrem key value1 value2 … 删除指定元素
- zremrangebyrank key start stop:按照排名范围删除元素。
- zremrangebyscore key min max:按照分数范围删除元素。
- 查询元素:
- zrange key start end withscores:查询start到end之间的成员。
- zrevrange key start end withscores:查询成员分数从大到小顺序的索引 start 到 end 的所有成员。
- zrangebyscore key min max withscores limit offset count:返回分数 min 到 max 的成员并按照分数从小到大排序, limit 是从 offset 开始展示几个元素。
4.2、Redis 实现方式
使用sortedset,用时间戳作为score,使用zadd key score1 value1
命令生产消息,使用zrangebysocre key min max withscores limit 0 1消费消息最早的一条消息。
这里选用 Redis 主要的原因就是其支持高性能的 score 排序,同时 Redis 的持久化 bgsave 特性,保证了消息的消费和存贮问题。bgsave 的原理是 fork 和 cow。fork 是指 Redis 通过创建子进程来进行 bgsave 操作, cow 指的是copy on write, 子进程创建后, 父进程通过共享数据段, 父进程继续提供读写服务, 写脏的页面数据会逐渐和子进程分离开来。
4.3、ACK
队列最重要的就是保证消息被成功消费,这里也不可避免的需要考虑这个问题。
- RabbitMQ 的 ACK机制:
- Publisher 把消息通知给 Consumer,如果 Consumer 已处理完任务,那么它将向 Broker 发送 ACK 消息,告知某条消息已被成功处理,可以从队列中移除。如果 Consumer 没有发送回 ACK 消息,那么 Broker 会认为消息处理失败,会将此消息及后续消息分发给其他 Consumer 进行处理 ( redeliver flag 置为 true )。
- 这种确认机制和 TCP/IP 协议确立连接类似。不同的是,TCP/IP 确立连接需要经过三次握手,而 RabbitMQ 只需要一次 ACK。
- 还有一个重要的是,RabbitMQ 当且仅当检测到 ACK 消息未发出且 Consumer 的连接终止时才会将消息重新分发给其他 Consumer ,因此不需要担心消息处理时间过长而被重新分发的情况。
- Redis 实现 ACK
- 需要在业务代码中处理消息失败的情况,回滚消息到原始等待队列。
- Consumer 挂掉,仍然需要回滚消息到等待队列中。
前者只需要在业务中处理消费异常的情况,后者则需要维护两个队列。
- Redis ACK 实现方案
- 维护一个消息记录表,存贮消息的消费记录,用于失败时回滚消息。表中记录消息ID、消息内容、消息时间、消息状态。
- 定时任务轮询该消息表,处理消费记录表中消费状态未成功的记录,重新放入等待队列。
4.4、多实例问题
多实例是指同一个服务部署在不同的地方,发挥相同的作用,此时就会导致同时消费同一个消息的问题。
一般情况下解决此类问题就需要考虑接入外部应用的辅助。常见的分布式锁的方案有:基于数据库实现分布式锁、基于缓存实现分布式锁、基于 Zookeeper 实现分布式锁,这里使用缓存也就是 Redis 解决问题。
- 利用 Redis 的 setnx 的互斥特性,把 key 当作锁存在 Redis 中,但是用 setnx 需要解决死锁和正确解锁的问题。
- 死锁:设置 key-value 的过期时间,并且使用 lua 脚本保证加锁和设置过期时间的原子性。
- 解锁:解锁需要保证是加锁客户端进行解锁操作。将 value 设置为 UUID,用对应的 UUID 去解锁保证是加锁客户端进行对应的解锁操作。
- 利用 Redis 的 List 实现一个 Publisher 推送消费保证只被消费一次,这种不用考虑死锁问题,但是需要额外维护一个队列。
五、总结
使用 Redis 实现的队列具有很好的扩展性,可以很便捷的应对需求的变更和业务的扩展,但是对于简单的场景直接使用定时任务会更加容易。在有大量的定时任务需要实现的时候,就可以考虑使用延迟队列去实现,让代码更具有扩展性。
Redis 作为消息队列的局限性很大,实现 ack 机制的成本相对较高,然而他的轻量级的特性以及兼容很多的数据结构,Redis 成熟的分布式、持久化、集群等技术体系,让他可以实现一些轻量级的队列。总之没有最好的技术,只有最好的 developer。
参考
5.RabbitMQ(三)RabbitMQ消息过期时间(TTL)
出处:https://zhuanlan.zhihu.com/p/87113913
=================================================================================================================================================
另一篇比较好的博文
在工作中想实现一个延迟功能,一般会借助rocketmq或者kafka的延迟队列功能来实现,但是这俩个消息中间件都有一个弊端,就是很难支持任意时间段的延迟,所以我想借助redis实现一个任意时间段的延迟功能
总体架构图
![](http://upload-images.jianshu.io/upload_images/8068221-caf8a1eabc30fe1d.png?imageMogr2/auto-orient/strip|imageView2/2/w/876/format/webp)
上述图主要分为5个模块
1 路由模块,为了支持分布式部署,每接收一个延迟消息,都为这个消息生成一个全局唯一的消息ID,根据消息ID和路由算法,决定把延迟消息的消息ID加入到对应的redis.sortSet队列中
2 消息存储,所有的消息元数据存储采用redis.hashmap结构,key为生成的全局消息ID(可以加一个前缀),value为消息的JSON格式,例如{"ttl":600,"topic":"XXX".......}
3 延迟队列,所有的消息的延迟存储在redis.sortSet中,sortSet中的每一个对象为全局生成的消息ID,score为到期时间时间戳
4 定时扫描timer,轮训redis.sortSet队列,使用ZRANGEBYSCORE命令,获取score小于等于当前时间的所有消息ID,然后根据消息ID查询redis.hashmap中的消息元数据,然后根据业务发送到对应的topic下
5 消息中间件,借助消息中间件,订阅者直接消费消息,达到延迟的功能
总结
上述只是整体的罗列了一下借助redis怎么实现任意时间段延迟的功能,一些细节没有详细说明,如果想实现一个比较完美的延迟功能,需要考虑以下几点
a 消息的发送失败如果处理
b redis操作没有事物保证
c 怎么保证hashmap和sortSet中的数据一致性
d 如果消息量大了怎么进行动态的扩展