前置知识
-
以RocketMQ为例
-
使用IDR启动RocketMQ,参考Eclpse启动RocketMq
-
大多数消息中间件都支持固定延时队列,比如RocketMQ支持的默认延时等级messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h。参考RocketMQ延时配置
-
Netty和Kafka中用到了时间轮的算法,自行百度
实现思路
- 配置MQ的延时等级为1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048, 4096, 8192,16384, 32768, 65536, 131072,单位s
- 对应1~18延时等级
- 当发送一个任意延时的消息时,把消息丢到固定等级的延时队列去,保留剩余延时时间到消息属性中;消费时检查剩余延时时间,如果大于0,继续丢回延时队列,重复上诉步骤直至剩余延时为0,此时真正的消费这条消息
- 举例,需要发送27s延时的消息,16<27<64,于是先丢到固定延时16s的队列中,然后设置剩余延时为11s,16s后这个消息被消费,检查剩余延时,此时8<11<16,再丢到8s延时队列中,重复上诉步骤直至剩余延时为0,把消息交给真正的消费中
注意
- 延时等级的设置可以使用1)2的倍数,2)斐波那契数列,3)1~59s+1~59m+1-23h+1-29d等形式的一百多个等级,时间轮降级会很快
- 延时最大的等级最好是一个比较大的值,比如支持最大30天,如果最大的是1天,那么30天的延时需要投递30次才会降级,所以最后的三个等级设置为1天,7天,15天比较合理