利用 RabbitMQ 死信队列和 TTL 实现定时任务

不要想象自己说的每句话，都是真理，但要保证自己说的每句话都是真话。—— 张杰

实际业务中对于定时任务的需求是不可避免的，例如，订单超时自动取消、每天定时拉取数据等，在 Node.js 中系统层面提供了 setTimeout、setInterval 两个 API 或通过 node-schedule 这种第三方库来实现。

作者简介：五月君，Nodejs Developer，慕课网认证作者，热爱技术、喜欢分享的 90 后青年，欢迎关注 Nodejs技术栈 和 Github 开源项目 https://www.nodejs.red

通过系统层面提供的 API 这种方式实现对于简单的定时任务是 ok 的，过于复杂的、可用性要求较高的系统就会存在以下缺点。

存在的一些问题

• 消耗系统内存，如果定时任务很多，长时间得不到释放，将会一直占用系统进程耗费内存。

• 单线程如何保障出现系统崩溃后之前的定时任务不受影响？多进程集群模式下一致性的保证？

• setTimeout、setInterval 会存在时间误差，对于时间精度要求较高的是不行的。

RabbitMQ TTL+DLX 实现定时任务

RabbitMQ 本身是不支持的，可以通过它提供的两个特性 Time-To-Live and Expiration、Dead Letter Exchanges 来实现，通过以下泳道图可以看到一个消息从发布到消费的整个过程。

死信队列

死信队列全称 Dead-Letter-Exchange 简称 DLX 是 RabbitMQ 中交换器的一种类型，消息在一段时间之后没有被消费就会变成死信被重新 publish 到另一个 DLX 交换器队列中，因此称为死信队列。

死信队列产生几种情况

• 消息被拒绝

• 消息 TTL 过期

• 队列达到最大长度

设置DLX的两个参数：

• deadLetterExchange: 设置 DLX，当正常队列的消息成为死信后会被路由到 DLX 中

• deadLetterRoutingKey: 设置 DLX 指定的路由键

注意：Dead-Letter-Exchange 也是一种普通的 Exchange

消息 TTL

消息的 TTL 指的是消息的存活时间，RabbitMQ 支持消息、队列两种方式设置 TTL，分别如下：

消息设置 TTL：对消息的设置是在发送时进行 TTL 设置，通过 x-message-ttl 或 expiration 字段设置，单位为毫秒，代表消息的过期时间，每条消息的 TTL 可不同。

队列设置 TTL：对队列的设置是在消息入队列时计算，通过 x-expires 设置，队列中的所有消息都有相同的过期时间，当超过了队列的超时设置，消息会自动的清除。

注意：如果以上两种方式都做了设置，消息的 TTL 则以两者之中最小的那个为准。

延迟队列实现 Node.js 版

推荐采用 amqplib 库，一个 Node.js 实现的 RabbitMQ 客户端。

初始化 RabbitMQ

1. // rabbitmq.js

2. // npm install amqplib

3. const amqp = require('amqplib');

4. let connection = null;

5. module.exports = {

6. connection,

7. init: () => amqp.connect('amqp://localhost:5672').then(conn => {

8. connection = conn;

9. console.log('rabbitmq connect success');

10. return connection;

11. })

12. }

生产者

1. /**

2. * 路由一个死信队列

3. * @param { Object } connnection

4. */

5. async function producerDLX(connnection) {

6. const testExchange = 'testEx';

7. const testQueue = 'testQu';

8. const testExchangeDLX = 'testExDLX';

9. const testRoutingKeyDLX = 'testRoutingKeyDLX';

10. const ch = await connnection.createChannel();

11. await ch.assertExchange(testExchange, 'direct', { durable: true });

12. const queueResult = await ch.assertQueue(testQueue, {

13. exclusive: false,

14. deadLetterExchange: testExchangeDLX,

15. deadLetterRoutingKey: testRoutingKeyDLX,

16. });

17. await ch.bindQueue(queueResult.queue, testExchange);

18. const msg = 'hello world!';

19. console.log('producer msg：', msg);

20. await ch.sendToQueue(queueResult.queue, new Buffer(msg), {

21. expiration: '10000'

22. });

23. ch.close();

24. }

消费者

1. // consumer.js

2. const rabbitmq = require('./rabbitmq.js');

3. /**

4. * 消费一个死信队列

5. * @param { Object } connnection

6. */

7. async function consumerDLX(connnection) {

8. const testExchangeDLX = 'testExDLX';

9. const testRoutingKeyDLX = 'testRoutingKeyDLX';

10. const testQueueDLX = 'testQueueDLX';

11. const ch = await connnection.createChannel();

12. await ch.assertExchange(testExchangeDLX, 'direct', { durable: true });

13. const queueResult = await ch.assertQueue(testQueueDLX, {

14. exclusive: false,

15. });

16. await ch.bindQueue(queueResult.queue, testExchangeDLX, testRoutingKeyDLX);

17. await ch.consume(queueResult.queue, msg => {

18. console.log('consumer msg：', msg.content.toString());

19. }, { noAck: true });

20. }

21. // 消费消息

22. rabbitmq.init().then(connection => consumerDLX(connection));

运行查看

分别执行消费者和生产者，可以看到 producer 在 44 秒发布了消息，consumer 是在 54 秒接收到的消息，实现了定时 10 秒种执行

1. $ node consumer # 执行消费者

2. [2019-05-07T08:45:23.099] [INFO] default - rabbitmq connect success

3. [2019-05-07T08:45:54.562] [INFO] default - consumer msg： hello world!

1. $ node producer # 执行生产者

2. [2019-05-07T08:45:43.973] [INFO] default - rabbitmq connect success

3. [2019-05-07T08:45:44.000] [INFO] default - producer msg： hello world!

管理控制台查看

testQu 队列为我们定义的正常队列消息过期，会变成死信，会被路由到 testQueueDLX 队列，形成一个死信队列。

注意问题

一个队列里的某个消息即使比同队列中的其它消息提前过期，也不会优先进入到死信队列，只有当过期的消息到了队列的顶端，才会被真正的丢弃或者进入死信队列。

源码地址

1. https://github.com/Q-Angelo/project-training/tree/master/nodejs/rabbitmq-timed-task

延迟队列实现 Java 版

构建生产者

生产者主要变动为设置消息过期时间

1. AMQP.BasicProperties basicProperties = new AMQP.BasicProperties.Builder()

2. .deliveryMode(2)

3. .contentEncoding("UTF-8")

4. .expiration("5000") // 设置 5 秒中过期

5. .build();

6. channel.basicPublish(dlxExchangeName, dlxRoutingKey, basicProperties, dlxMsg.getBytes());

构建消费者

核心代码变动如下，除了绑定一个正常的交换机、队列之外，还需在绑定一个死信的交换机、队列

1. public static void main(String[] args) throws Exception {

2. ...

3. // 定义正常交换机、队列等信息

4. String dlxExchangeName = "dlx_exchange_name";

5. String exchangeType = "direct";

6. String dlxRoutingKey = "dlx_routingKey";

7. String dlxQueueName = "dlx_queue";

8. // 定义死信队列交换机、队列等信息

9. String dlxTestExchangeName = "dlx_test_exchange_name";

10. String dlxTestRoutingKey = "dlx_test_routingKey";

11. String dlxTestQueueName = "dlx_test_queue";

12. // 声明一个正常的交换机、队列和绑定关系

13. channel.exchangeDeclare(dlxExchangeName, exchangeType, true, false, null);

14. // 声明死信队列交换机

15. Map<String, Object> arguments = new HashMap<String, Object>();

16. arguments.put("x-dead-letter-exchange", dlxTestExchangeName);

17. channel.queueDeclare(dlxQueueName, true, false, false, arguments);

18. channel.queueBind(dlxQueueName, dlxExchangeName, dlxRoutingKey);

19. // 死信队列的交换机、队列声明和绑定关系

20. channel.exchangeDeclare(dlxTestExchangeName, "direct", true, false, null);

21. channel.queueDeclare(dlxTestQueueName, true, false, false, null);

22. channel.queueBind(dlxTestQueueName, dlxTestExchangeName, dlxRoutingKey);

23. ...

24. // 6. 设置 channel

25. channel.basicConsume(dlxTestQueueName, true, consumer);

26. System.out.println("消费端启动成功！");

27. }

源码地址

1. https://github.com/Q-Angelo/SpringBoot-Course/tree/master/chapter8/chapter8-1/src/main/java/com/may/rabbitmq/dlx/helloworld

总结

延迟队列在现实业务场景中，还是有很多实际用途的，订单的超时取消、重试等，都可以借助此来完成，通过本文希望你能掌握什么是延迟队列，在 RabbitMQ 中的实现主要是根据 TTL + 死信队列来完成的，本文最后采用了 Node.js 和 Java 分别进行了实践，希望能帮助到你。

▼

往期精彩回顾

▼

在看点这里