Node.js结合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
rabbitmq.js
// npm install amqplibconst amqp = require('amqplib');
let connection = null;
module.exports = { connection,
init: () => amqp.connect('amqp://localhost:5672').then(conn => { connection = conn;
console.log('rabbitmq connect success');
return connection; })}
- 生产者
/** * 路由一个死信队列 * @param { Object } connnection */async function producerDLX(connnection) { const testExchange = 'testEx'; const testQueue = 'testQu'; const testExchangeDLX = 'testExDLX'; const testRoutingKeyDLX = 'testRoutingKeyDLX';
const ch = await connnection.createChannel(); await ch.assertExchange(testExchange, 'direct', { durable: true }); const queueResult = await ch.assertQueue(testQueue, { exclusive: false, deadLetterExchange: testExchangeDLX, deadLetterRoutingKey: testRoutingKeyDLX, }); await ch.bindQueue(queueResult.queue, testExchange); const msg = 'hello world!'; console.log('producer msg:', msg); await ch.sendToQueue(queueResult.queue, new Buffer(msg), { expiration: '10000' });
ch.close();}
- 消费者
consumer.js
const rabbitmq = require('./rabbitmq.js');
/** * 消费一个死信队列 * @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 => { console.log('consumer msg:', msg.content.toString()); }, { noAck: true });}
// 消费消息rabbitmq.init().then(connection => consumerDLX(connection));
- 运行查看
分别执行消费者和生产者,可以看到 producer 在44秒发布了消息,consumer 是在54秒接收到的消息,实现了定时10秒种执行
$ node consumer # 执行消费者[2019-05-07T08:45:23.099] [INFO] default - rabbitmq connect success[2019-05-07T08:45:54.562] [INFO] default - consumer msg: hello world!
$ node producer # 执行生产者[2019-05-07T08:45:43.973] [INFO] default - rabbitmq connect success[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
- React Native 实现基于react-native-tab-navigator库Tab切换封装
- 16.4 配置Tomcat监听80端口
- JDK容器学习之Queue:LinkedBlockingQueue
- Linux基础(day59)
- 16.3 安装Tomcat
- 16.2 安装jdk
- UITabBarController实现Tab切换
- React Native库版本升级与降级
- Java并发学习之Volatile及内存模型探究
- Java并发学习之CountDownLatch实现原理及使用姿势
- Linux基础(day58)
- 携程Android App插件化和动态加载实践
- 15.5 使用pure-ftpd搭建ftp服务
- JDK容器学习之Queue: ArrayBlockingQueue
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法