Node.js结合RabbitMQ延迟队列实现定时任务

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

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

存在的一些问题

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

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

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

RabbitMQ TTL+DLX 实现定时任务

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

死信队列

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

死信队列产生几种情况

1. 消息被拒绝

2. 消息TTL过期

3. 队列达到最大长度

设置DLX的两个参数：

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

2. deadLetterRoutingKey: 设置DLX指定的路由键

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

消息TTL

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

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

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

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

Nodejs操作RabbitMQ实现延迟队列

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

• 初始化RabbitMQ

rabbitmq.js

1. // npm install amqplib

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

3. let connection = null;

4. module.exports = {

5. connection,

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

7. connection = conn;

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

9. return connection;

10. })

11. }

• 生产者

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. }

• 消费者

consumer.js

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

2. /**

3. * 消费一个死信队列

4. * @param { Object } connnection

5. */

6. async function consumerDLX(connnection) {

7. const testExchangeDLX = 'testExDLX';

8. const testRoutingKeyDLX = 'testRoutingKeyDLX';

9. const testQueueDLX = 'testQueueDLX';

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

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

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

13. exclusive: false,

14. });

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

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

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

18. }, { noAck: true });

19. }

20. // 消费消息

21. 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 队列，形成一个死信队列。

• 源码地址：RabbitMQ延迟队列实现定时任务（Node.js客户端版Demo）

作者：五月君
链接：https://www.imooc.com/article/286402
来源：慕课网
Node.js技术栈: https://github.com/Q-Angelo/Nodejs-Roadmap