Node.js結合RabbitMQ延遲隊列實現定時任務

時間 2019-11-30

標籤 node.js node 結合 rabbitmq 延遲隊列實現定時任務欄目 Node.js 简体版

原文原文鏈接

實際業務中對於定時任務的需求是不可避免的，例如，訂單超時自動取消、天天定時拉取數據等，在Node.js中系統層面提供了setTimeout、setInterval兩個API或經過node-schedule這種第三方庫來實現。html

經過這種方式實現對於簡單的定時任務是ok的，過於複雜的、可用性要求較高的系統就會存在如下缺點。node

存在的一些問題git
1. 消耗系統內存，若是定時任務不少，長時間得不到釋放，將會一直佔用系統進程耗費內存。
2. 單線程如何保障出現系統崩潰後以前的定時任務不受影響？多進程集羣模式下一致性的保證？
3. setTimeout、setInterval會存在時間偏差，對於時間精度要求較高的是不行的。
RabbitMQ TTL+DLX 實現定時任務github

RabbitMQ自己是不支持的，能夠經過它提供的兩個特性Time-To-Live and Expiration、Dead Letter Exchanges來實現，經過如下泳道圖能夠看到一個消息從發佈到消費的整個過程。shell

死信隊列

死信隊列全稱 Dead-Letter-Exchange 簡稱 DLX 是 RabbitMQ 中交換器的一種類型，消息在一段時間以後沒有被消費就會變成死信被從新 publish 到另外一個 DLX 交換器隊列中，所以稱爲死信隊列。npm

死信隊列產生幾種狀況bash
- 消息被拒絕
- 消息TTL過時
- 隊列達到最大長度
設置DLX的兩個參數：async
- deadLetterExchange: 設置DLX，當正常隊列的消息成爲死信後會被路由到DLX中
- deadLetterRoutingKey: 設置DLX指定的路由鍵

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

消息TTL

消息的TTL指的是消息的存活時間，RabbitMQ支持消息、隊列兩種方式設置TTL，分別以下：spa

消息設置TTL：對消息的設置是在發送時進行TTL設置，經過x-message-ttl 或expiration 字段設置，單位爲毫秒，表明消息的過時時間，每條消息的TTL可不一樣。
隊列設置TTL：對隊列的設置是在消息入隊列時計算，經過 x-expires 設置，隊列中的全部消息都有相同的過時時間，當超過了隊列的超時設置，消息會自動的清除。

注意：若是以上兩種方式都作了設置，消息的TTL則以二者之中最小的那個爲準。

Nodejs操做RabbitMQ實現延遲隊列

推薦採用 amqplib庫，一個Node.js實現的RabbitMQ客戶端。

初始化RabbitMQ

rabbitmq.js

// npm install amqplib
const 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!
複製代碼