rabbitmq 延时队列实现定时任务
Posted xiaosongjiang
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了rabbitmq 延时队列实现定时任务相关的知识,希望对你有一定的参考价值。
场景
实际业务中对于定时任务的需求是不可避免的,例如,订单超时自动取消、每天定时拉取数据等,在Node.js中系统层面提供了setTimeout、setInterval两个API或通过node-schedule这种第三方库来实现。通过这种方式实现对于简单的定时任务是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则以两者之中最小的那个为准。
Nodejs操作RabbitMQ实现延迟队列
推荐采用 amqplib库,一个Node.js实现的RabbitMQ客户端。
初始化RabbitMQ
const amqp = require(‘amqplib‘); const log4js = require(‘log4js‘); const logger = log4js.getLogger(); logger.level = ‘info‘; module.exports = { logger, init: () => amqp.connect(‘amqp://122.51.9.11:5672‘).then((connection) => { logger.info(‘rabbitmq connect success‘); return connection; }), };
生产者
const random = require(‘string-random‘); const rabbitmq = require(‘./rabbitmq.js‘); const logger = rabbitmq.logger; const sleep = (time) => new Promise((resolve) => setTimeout(resolve, time)); async function producerDLX(connnection) { const testQueue = ‘testQu‘; const testExchange = ‘testEx‘; const testRoutingKey = ‘testRoutingKey‘; const testExchangeDLX = ‘testExDLX‘; const testRoutingKeyDLX = ‘testRoutingKeyDLX‘; const msg = ‘Producer‘; const ch = await connnection.createChannel(); await ch.assertExchange(testExchange, ‘direct‘, { durable: true }); const queueResult = await ch.assertQueue(testQueue, { exclusive: false, messageTtl: 10000, deadLetterExchange: testExchangeDLX, deadLetterRoutingKey: testRoutingKeyDLX, }); await ch.bindQueue(queueResult.queue, testExchange, testRoutingKey); for (let i = 0; i < 5; i++) { await sleep(2000); const cMsg = `${i}:${msg} 消息 =>${random(10)}`; logger.info(cMsg); await ch.publish(testExchange, testRoutingKey, Buffer.from(cMsg)); } await ch.close(); } // 发送消息 rabbitmq.init().then((connection) => producerDLX(connection));
消费者
const rabbitmq = require(‘./rabbitmq.js‘); const logger = rabbitmq.logger; /** * 消费一个死信队列 * @param { Object } connnection */ async function consumerDLX(connnection) { const testExchangeDLX = ‘testExDLX‘; const testRoutingKeyDLX = ‘testRoutingKeyDLX‘; const testQueueDLX = ‘testQueueDLX‘; const ch = await connnection.createChannel(); await ch.assertExchange(testExchangeDLX, ‘direct‘, { durable: true }); const queueResult = await ch.assertQueue(testQueueDLX, { exclusive: false, }); await ch.bindQueue(queueResult.queue, testExchangeDLX, testRoutingKeyDLX); await ch.consume( queueResult.queue, (msg) => { logger.info(‘consumer msg:‘, msg.content.toString()); }, { noAck: true } ); } // 消费消息 rabbitmq.init().then((connection) => consumerDLX(connection));
分别执行消费者和生产者,可以看到 producer 在45秒发布了消息,consumer 是在55秒接收到的消息,实现了定时10秒种执行
参考
以上是关于rabbitmq 延时队列实现定时任务的主要内容,如果未能解决你的问题,请参考以下文章