常用方式
- Redission
- RabbitMQ插件
- RabbitMQ私信队列
- RocketMQ延迟消息
- Redist的zset
- Redis的过期监听
- kafka时间轮
- 定时任务
- Netty的时间轮
- JDK自带的DelayQueue (延迟队列)
- 被动关闭
不同的场景中也适合不同的方案:
- 自己玩玩:被动关闭
- 单体应用,业务量不大:Netty的时间轮、JDK自带的DelayQueue、定时任务
- 分布式应用,业务量不大:Redis过期监听、RabbitMQ死信队列、Redis的zset、定时任务
- 分布式应用,业务量大、并发高:Redission、RabbitMQ插件、kafka时间轮、RocketMQ延迟消息
Redission
Redisson是一个在Redis的基础上实现的框架,它不仅提供了一系列的分布式的Java常用对象,还提供了许多分布式服务。
Redission中定义了分布式延迟队列RDelayedQueue,这是一种基于我们前面介绍过的zset结构实现的延时队列,它允许以指定的延迟时长将元素放到目标队列中。
其实就是在zset的基础上增加了一个基于内存的延迟队列。当我们要添加一个数据到延迟队列的时候,redission会把数据+超时时间放到zset中,并且起一个延时任务,当任务到期的时候,再去zset中把数据取出来,返回给客户端使用。
大致思路就是这样的,感兴趣的大家可以看一看RDelayedQueue的具体实现。
RabbitMQ插件
其实,基于RabbitMQ的话,可以不用死信队列也能实现延迟消息,那就是基于rabbitmq_delayed_message_exchange插件,这种方案能够解决通过死信队列实现延迟消息出现的消息阻塞问题。但是该插件从RabbitMQ的3.6.12开始支持的,所以对版本有要求。
官方插件:
但是基于插件的这种方式,消息并不会立即进入队列,而是先把他们保存在一个基于Erlang开发的Mnesia数据库中,然后通过一个定时器去查询需要被投递的消息,再把他们投递到x-delayed-message队列中。
基于RabbitMQ插件的方式可以实现延迟消息,并且不存在消息阻塞的问题,但是因为是基于插件的,而这个插件支持的最大延长时间是(2^32)-1 毫秒,大约49天,超过这个时间就会被立即消费。但是他基于RabbitMQ实现,所以在可用性、性能方便都很不错
RocketMQ延迟消息
相比于Kafka来说,RocketMQ中有一个强大的功能,那就是支持延迟消息。
延迟消息,当消息写入到Broker后,不会立刻被消费者消费,需要等待指定的时长后才可被消费处理的消息,称为延时消息。
有了延迟消息,我们就可以在订单创建好之后,发送一个延迟消息,比如20分钟取消订单,那就发一个延迟20分钟的延迟消息,然后在20分钟之后,消息就会被消费者消费,消费者在接收到消息之后,去关单就行了。
但是,RocketMQ的延迟消息并不是支持任意时长的延迟的,它只支持:1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h这几个时长。(商业版支持任意时长)
可以看到,有了RocketMQ延迟消息之后,我们处理上就简单很多,只需要发消息,和接收消息就行了,系统之间完全解耦了。但是因为延迟消息的时长受到了限制,所以并不是很灵活。
如果我们的业务上,关单时长刚好和RocketMQ延迟消息支持的时长匹配的话,那么是可以基于RocketMQ延迟消息来实现的。否则,这种方式并不是最佳的。
kafka的时间轮
既然基于Netty的时间轮存在一些问题,那么有没有其他的时间轮的实现呢?
还真有的,那就是Kafka的时间轮,Kafka内部有很多延时性的操作,如延时生产,延时拉取,延时数据删除等,这些延时功能由内部的延时操作管理器来做专门的处理,其底层是采用时间轮实现的。
而且,为了解决有一些时间跨度大的延时任务,Kafka 还引入了层级时间轮,能更好控制时间粒度,可以应对更加复杂的定时任务处理场景;
Kafka 中的时间轮的实现是 TimingWheel 类,位于 kafka.utils.timer 包中。基于Kafka的时间轮同样可以得到O(1)时间复杂度,性能上还是不错的。
基于Kafka的时间轮的实现方式,在实现方式上有点复杂,需要依赖kafka,但是他的稳定性和性能都要更高一些,而且适合用在分布式场景中。