兔老大的系统设计（二）定时系统（延时队列）

public class Order implements Delayed {
    //延迟时间
    @JsonFormat(locale = "zh", timezone = "GMT+8", pattern = "yyyy-MM-dd HH:mm:ss")
    private long time;
    String name;

    public Order(String name, long time, TimeUnit unit) {
        this.name = name;
        this.time = System.currentTimeMillis() + (time &gt; 0 ? unit.toMillis(time) : 0);
    }

    @Override
    public long getDelay(TimeUnit unit) {
        return time - System.currentTimeMillis();
    }
    @Override
    public int compareTo(Delayed o) {
        Order Order = (Order) o;
        long diff = this.time - Order.time;
        if (diff <= 0) {
            return -1;
        } else {
            return 1;
        }
    }
}

DelayQueue 的 put 方法是线程安全的，内部使用了ReentrantLoc k进行线程同步。

上边只是简单的实现，实际开发中会有专门的线程负责消息的入队与消费。

4.1.3 分析

分析：事实上，如无必要，我们应该尽可能使用语言自带的库，而非过度设计。从这方面考虑，DelayQueue无疑是一个简单优秀的实现，但是在大型项目中，本地存储的方案确实不太适用。

无论如何，我们仍然可以对它大根堆和线程控制的方法进行学习和借鉴。

4.2 Rab bitMQ

Rab bitMQ 本身并不直接提供对延迟队列的支持，我们依靠 Rab bitMQ 的TTL以及死信队列功能，来实现延迟队列的效果。那就让我们首先来了解一下，Rab bitMQ 的死信队列以及 TTL 功能。

4.2.1死信队列

死信队列实际上是一种 Rab bitMQ 的消息处理机制，当 Rab bmitMQ 在生产和消费消息的时候，消息遇到如下的情况，就会变成“死信”：

消息被拒绝并且不再重新投递
消息超时未消费，也就是 TTL 过期了
消息队列到达最大长度

消息一旦变成一条死信，便会被重新投递到死信交换机（Dead-Letter-Ex change），然后死信交换机根据绑定规则转发到对应的死信队列上，监听该队列就可以让消息被重新消费。

4.2.2消息生存时间 TTL

TTL（Time-To-Live）表示了一条消息的最大生存时间，单位为毫秒。如果一条消息在 TTL 设置的时间内没有被消费，那么它就会变成一条死信，进入我们上面所说的死信队列。

如何设置消息的 TTL 属性

一种方式是直接在创建队列的时候设置整个队列的 TTL 过期时间，所有进入队列的消息，都被设置成了统一的过期时间，一旦消息过期，马上就会被丢弃，进入死信队列，在延迟队列的延迟时间为固定值的时候，比较适合使用这种方式：

Map<String, Object> args = new HashMap<String, Object>();
args.put("x-message-ttl", 6000);
channel.queueDeclare(queueName, durable, exclusive, autoDelete, args);

另一种方式是针对单条消息设置，参考代码如下，该消息被设置了 6 秒的过期时间：

AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
builder.expiration("6000");
AMQP.BasicProperties properties = builder.build();
channel.basicPublish(exchangeName, routingKey, mandatory, properties, "msg content".getBytes());

4.2.3实现延迟队列

我们利用死信队列的这个属性，把需要延迟的消息，将 TTL 设置为其延迟时间，投递到 RabbitMQ 的普通队列中，一直不去消费它，那么经过 TTL 的时间后，消息就会自动被投递到死信队列，这时候我们使用消费者进程实时地去消费死信队列中的消息，就实现了延迟队列的效果。

从下图可以直观的看出使用 RabbitMQ 实现延迟队列的整体流程：

使用 RabbitMQ 来实现延迟队列，具有很明显的一些优势：比如消息可靠发送、消息可靠投递、死信队列来保障消息至少被消费一次以及未被正确处理的消息不会被丢弃。另外，通过 RabbitMQ 集群的特性，可以很好的解决单点故障问题，不会因为单个节点挂掉导致延迟队列不可用或者消息丢失。

4.2.4存在的bug

RabbitMQ 在检查消息是否过期时，只会检查第一个消息是否过期，并不会校验后面消息过期的情况比如第一个消息设置了 20s 的 TTL，第二个消息设置了 10s 的 TTL，那么 RabbitMQ 会等到第一个消息过期之后，才会让第二个消息过期。

解决这个问题的方法也很简单，只需要安装 RabbitMQ 的一个插件即可：

Community Plugins — RabbitMQ

安装好这个插件后，所有的消息就都能按照被设置的 TTL 过期了。

4.2.5插件实现原理

上面使用 DLX + TTL 的模式，消息首先会路由到一个正常的队列，根据设置的 TTL 进入死信队列，与之不同的是通过 x-delayed-message 声明的交换机，它的消息在发布之后不会立即进入队列，先将消息保存至 Mnesia。

这个插件将会尝试确认消息是否过期，首先要确保消息的延迟范围是 Delay > 0, Delay =<ERL_MAX_T（在 Erlang 中可以被设置的范围为 (2^32)-1 毫秒），如果消息过期通过 x-delayed-type 类型标记的交换机投递至目标队列，整个消息的投递过程也就完成了。

4.2.6 分析

作为网上流传非常广的一种方案，它似乎真的是一种不错的实现。

尤其是，由于 RabbitMQ 本身的消息可靠发送、消息可靠投递、死信队列等特性，可以保障消息至少被消费一次以及未被正确处理的消息不会被丢弃，让消息的可靠性有了保障。

但是这方案有如下缺点：

1、为了解决一个问题，又引入了队列交换机+mq+私信队列交换机+私信队列+插件，我们并不希望引入如此复杂不可控的架构。

2、配置麻烦，额外增加死信交换机和死信队列等配置，不好维护

3、不可靠，实际测试环境延时插件有时收不到消息，不是很稳定。配置错误、生产者消费者连接的队列错误和其他未知因素都有可能造成延迟失效。

4、真实消费原因不唯一：消息被拒绝、消息过期、消息超长等等原因都会进入死信队列，这种不唯一也是我们无法忍受的。，我们无法知道死信队列中是否都是过期消息。

4.3 kafka-TimeWheel

TimeWheel 时间轮算法，是一种实现延迟队列的巧妙且高效的算法，被应用在 Netty，Zookeeper，Kafka 等各种框架中。

4.3.1时间轮

如上图所示，时间轮是一个存储延迟消息的环形队列，其底层采用数组实现，可以高效循环遍历。这个环形队列中的每个元素对应一个延迟任务列表，这个列表是一个双向环形链表，链表中每一项都代表一个需要执行的延迟任务。时间轮会有表盘指针，表示时间轮当前所指时间，随着时间推移，该指针会不断前进，并处理对应位置上的延迟任务列表。

4.3.2添加延迟任务

由于时间轮的大小固定，并且时间轮中每个元素都是一个双向环形链表，我们可以在O(1) 的时间复杂度下向时间轮中添加延迟任务。

如下图，例如我们有一个这样的时间轮，在表盘指针指向当前时间为 2 时，我们需要新添加一个延迟 3 秒的任务，我们可以快速计算出延迟任务在时间轮中所对应的位置为 5，并添加到位置 5 上任务列表尾部。

4.3.3多层时间轮

上面的时间轮的大小是固定的，只有 12 秒。如果此时我们有一个需要延迟 200 秒的任务，我们应该怎么处理呢？直接扩充整个时间轮的大小吗？这显然不可取，因为这样做的话我们就需要维护一个非常非常大的时间轮，内存是不可接受的，而且底层数组大了之后寻址效率也会降低，影响性能。

为此，Kafka 引入了多层时间轮的概念。其实多层时间轮的概念和我们的机械表上时针、分针、秒针的概念非常类似，当仅使用秒针无法表示当前时间时，就使用分针结合秒针一起表示。同样的，当任务的到期时间超过了当前时间轮所表示的时间范围时，就会尝试添加到上层时间轮中，如下图所示：

第一层时间轮整个时间轮所表示时间范围是 0-12 秒，第二层时间轮每格能表示的时间范围是整个第一层时间轮所表示的范围也就是 12 秒，所以整个第二层时间轮能表示的时间范围即 12*12=144 秒，依次类推第三层时间轮能表示的范围是 1728 秒，第四层为 20736 秒等等。

比如现在我们需要添加一个延时为 200 秒的延迟消息，我们发现其已经超过了第一层时间轮能表示的时间范围，我们就需要继续往上层时间轮看，将其添加在第二层时间轮 200/12 = 17 的位置，然后我们发现 17 也超过了第二次时间轮的表示范围，那么我们就需要继续往上层看，将其添加在第三层时间轮的 17/12 = 2 的位置。

4.3.4具体实现

Kafka 中时间轮算法添加延迟任务以及推动时间轮滚动的核心流程如下，其中 Bucket 即时间轮中的延迟任务队列，并且 Kafka 引入的 DelayQueue 解决了多数 Bucket 为空导致的时间轮滚动效率低下的问题：

使用时间轮实现的延迟队列，能够支持大量任务的高效触发。并且在 Kafka 的时间轮算法的实现方案中，还引入了 DelayQueue，使用 DelayQueue 来推送时间轮滚动，而延迟任务的添加与删除操作都放在时间轮中，这样的设计大幅提升了整个延迟队列的执行效率。

4.3.5 分析

从复杂度上说：相比 DelayQueue ，时间轮在复杂度上有优势。DelayQueue 由于涉及到调整数据的位置，插入和移除复杂度是 O(lgn)，而时间轮在插入和移除的复杂度都是 O(1)。

从实际上说：相比其它MQ，kafka在我的认知里是最优秀的，事实上在我的十万级压测中，它是唯一性能达标的MQ（有些MQ已经接近挂了它还很健康）。同时kafka也有一定的持久化方案。

但是这种方案依旧有一些问题：

1、正如我开头提到的，需求很可能是保存一个月甚至更长时间，超过了默认的log.retention.hours（168）的大小。

2、我们希望执行时间视可修改的，但是kafka的消息一旦由生产者发送，则不可变。关于这方面讨论我贴了一个链接，感兴趣的可以看看。stack overflow问题https://stack overflow.com/questions/60046428/what-is-kafka–message-tweaking

其实探索到想用众多MQ来实现延迟队列时，我越来越清晰的有一种感觉：非逼着众多MQ（比如 kafka）做不擅长的事情本身就有问题，人家的定位就是消息队列，而不是替你保存动辄一个月才执行的消息然后精准执行。

4.4 一些其他方案

这里我不准备继续分析所有方案的优缺点了，因为这是很无聊的（而且影响接下来方案叙述的节奏），如Quartz、ActiveMQ、RocketMQ、nsq、pulsar等等，原因无非是性能不达标、时间粒度不够、存储时间不够等等，在这里放一张MQ的对比图：（如果满足你的要求，当然也可以用）

4.5 Redis ZSet

基于4.3结尾的考虑，首先要有地方做持久化，redis作为no sql的老大，呼之欲出。

4.5.1过期回调

只是提一嘴，这种歪门邪道的实现就不要想了，事实上容易出大问题，有兴趣可以了解。

4.5.2 正解介绍

Redis有一个有序集合的数据结构ZSet，ZSet中每个元素都有一个对应Score，ZSet 中所有元素是按照其 Score 进行排序的。我们利用它Score有序的属性，可以对入队的成员，按过期时间从小到大排列！

那么我们可以通过以下这几个操作使用 Redis 的 ZSet 来实现一个延迟队列：

入队操作：我们将需要处理的任务，按其需要延迟处理时间作为 Score 加入到 ZSet 中。Redis 的 ZAdd 的时间复杂度是O(logN)，N是ZSet 中元素个数，因此我们能相对比较高效的进行入队操作。
起一个进程定时（比如每隔一秒）查询 ZSet 中 Score 最小的元素，查询结果有两种情况：

查询出的分数小于等于当前时间戳，说明到这个任务需要执行的时间了，则去异步处理该任务；
查询出的分数大于当前时间戳，说明 ZSet 中所有的任务都还没有到需要执行的时间，则休眠一秒后继续查询；

4.5.3 分析

从上述的讨论中我们可以看到，通过 Redis Zset 实现延迟队列是一种理解起来较为直观，可以快速落地的方案。并且我们可以依赖 Redis 自身的持久化来实现持久化，使用 Redis 集群来支持高并发和高可用，是一种不错的延迟队列的实现方案。

但是redis同样有缺点（但是被解决了）：

1、定位问题：和上文提到的众多mq一样，redis的定位并不是延迟队列。

（经验告诉我们，如果硬要用与需求定位不符的东西，就是容易出问题）

但是由于这种方式实在过于简单好用，在业界确实非常太有市场，我记得redis 作者本人都曾经谈过这个问题，告诉大家最好不要把redis用作消息队列之类的，只不过貌似没人听。

2、持久化不是百分百可靠：redis 持久化两种方式我就不讲了，最高级的持久化配置就是每次操作都记录，但是由于性能问题，基本不可能这样配（事实上大公司有明确规定不能这样配）。

3、真实案例，如果qps过高，虽然redis扛得住（和kafka一样，真男人啊），但是我们的服务扛不住。

五、方案

5.1 思考

读者紧接着4.5.3的想法不要断，我们想用redis，但是有三个问题，如何解决呢？

问题2：没的说，上mysql万事大吉。

问题3：redis接MQ，完美解决问题，这不就是mq天天吹的其中一大作用吗。

问题1：我们只是用来排个序，消息队列和持久化都不是redis做了，符合定位。

5.2 整体 架构

基于这些考虑，最终我们的架构是这样的：

其他服务想使用定时功能，调用写接口，核心线路有两条：

第一条：

1）将延迟的消息任务通过 hash 算法路由至不同的 Redis Key 上

2）每个 Redis Key 都对应建立一个处理进程，称为 Event ，轮询，查询是否有待处理的延迟消息

3） Event 进程不负责业务，基本只负责分发消息，具体的业务逻辑通过kafka解耦，消费者实现。

4）我们规定消费者一定要上报执行结果，以便我们决定是否重复请求

第二条：

1）将消息写入DB（或更改）

2）event 进程扫过期一段时间的任务（可配置）

3）主动请求消费者执行

5.3 细节补充

1）mysql

表结构

2）redis

如觉得写mysql这条链路也太麻烦，并且没有存储很久的需求，可以用redis自身的持久化功能，同时开启RDB和AOF，AOF设置everysec，即每秒异步刷盘一次。极端情况下，可能会丢失一秒的数据。

高可用使用的是redis的主从复制模式。服务高可用方面，在实现过程中考虑了服务节点的横向扩展，Timer、Cleaner等对同一个redis队列的操作都加了分布式锁。每个服务节点都是无状态的，不需要进行元数据同步等操作，少数服务节点宕机不影响整个服务的可用性。

3）监控

对于消息堆积，以及消息超过重试次数被丢弃等场景，说明消费端服务异常，没有正常消费及ack，需要及时上报并通知给业务方及服务提供方，方便快速发现并排查问题。

4）mq

mq的选择上文有对比图，详细分析以后补充吧。

六、总结和QA

分析架构

可能很多人会有些疑惑：你抛出来看起来这么复杂的图，实现起来是不是很麻烦？它真的很好用吗？

下面首先回忆第三章，看是否实现目标。

从性能出发，整条链路看：

1）redis key可以增加，不用担心量大影响性能

2）event定时任务每秒轮询，基本没延迟

3）event没业务逻辑，校验+转发个消息很快

3）消息队列选用的性能最强的kafka

实际测试也符合要求

从可用出发，整条链路看：

1）将延迟的消息任务通过 ha sh 算法路由至不同的 Redis Key 上，这样做有两大好处：

避免了当一个 KEY 在存储了较多的延时消息后，入队操作以及查询操作速度变慢的问题（两个操作的时间复杂度均为O(logN)）。
系统具有了更好的横向可扩展性，当数据量激增时，我们可以通过增加Redis Key 的数量来快速的扩展整个系统。

但是会存在一个问题，因为增加key的数量，必然涉及到hash算法范围的调整，那么原先集合中的元素就不能通过新的hash算法路由到，所以需要采用一致性hash算法。

2）所有的 Event 进程只负责分发消息，具体的业务逻辑通过MQ解耦，由消费者异步处理，这么做的好处也是显而易见的：

Event 进程只负责分发消息，那么其处理消息的速度就会非常快，就不太会出现因为业务逻辑复杂而导致消息堆积的情况。
采用一个额外的消息队列后，消息处理的可扩展性也会更好，我们可以通过增加消费者进程数量来扩展整个系统的消息处理能力。
Event 是多机多进程模型，保证整个系统的高可用性。采用 Zookeeper 选主的方式，保证同一时间只会有一个进程去处理消息，一旦 Zookeeper 的 leader 主机宕机，会自动选择新的 leader 来处理。