Node.js連接RabbitMQ，斷線重連，動態綁定routing key

Lsnsh 發布于2019-07-01 17:31 / 2734人閱讀

摘要：官方提供的教程，是基于回調的。下面將給出基于式的寫法。并且實現動態的隊列綁定初始化配置地址交換機名稱讀取在跑多實例時，例如在中，可以獲取當前的名稱多實例時，寫日志，或者建立連接時，最好帶上名稱，如果出現問題，也比較好定位哪個出現的問題。

RabbitMQ官方提供的教程https://www.rabbitmq.com/tuto...，是基于回調的。

下面將給出基于Promise式的寫法。并且實現動態的隊列綁定

初始化配置

const amqp = require("amqplib")
// rabbitMQ地址
const {amqpAddrHost} = require("../config/index.js")

// 交換機名稱
const ex = "amq.topic"

const amqpAddr = `amqp://${amqpAddrHost}`

// 讀取HOSTNAME, 在跑多實例時，例如在k8s中，HOSTNAME可以獲取當前pod的名稱
// 多實例時，寫日志，或者建立連接時，最好帶上pod名稱，如果出現問題，也比較好定位哪個pod出現的問題。
const hostName = process.env.HOSTNAME

// 隊列的屬性設置
// 一般來說，最好設置隊列自動刪除autoDelete，當鏈接斷開時，隊列也會刪除，這樣不會產生非常多的無用隊列
// durable是用來的持久化的，最好也可以設置成不持久化

const queueAttr = {autoDelete: true, durable: false}

// 定義channel的引用，當鏈接建立時，所有方法都可以通過引用CH來獲取channel方法
let CH = null

向隊列發送消息的函數

// 向隊列發送消息的函數
function publishMessage (msg) {
  if (!CH) {
    return ""
  }

  msg = JSON.stringify(msg)
  // 指定交換機ex, routing key, 以及消息的內容
  CH.publish(ex, eventBusTopic, Buffer.from(msg))
}

當鏈接rabbitMQ斷開時，要主動去重連

function reconnectRabbitMq () {
  log.info("reconnect_rabbit_mq")
  connectRabbitMq()
}

連接rabbitMQ的主要函數

function connectRabbitMq () {
  amqp.connect(amqpAddr, {
    // 設置connection_name的屬性，可以在rabbitMQ的控制臺的UI上，看到連接是來自哪個實例
    clientProperties: {
      connection_name: hostName
    }
  })
  .then((conn) => {
    log.info("rabbitmq_connect_successd")
    // 一定要加上鏈接的報錯事件處理，否則一旦報error錯，如果不處理這個錯誤，程序就會崩潰
    // error是個特別的事件，務必要處理的
    // 報錯就直接去重連
    conn.on("error", (err) => {
      log.error("connect_error " + err.message, err)
      reconnectRabbitMq()
    })
    // 創建channel
    return conn.createChannel()
  })
  .then((ch) => {
    CH = ch
    // 初始化交換機
    ch.assertExchange(ex, "topic", {durable: true})
    // 初始化一個隊列，隊列名就用hostName, 比較容易從對列名上知道是哪個實例創建的隊列
    return ch.assertQueue(hostName, queueAttr)
  })
  .then((q) => {
    // 可以在隊列初始化完畢就立即綁定routing key, 也可以暫時不綁定，后續動態的綁定
    // CH.bindQueue(q.queue, ex, "some.topic.aaa")
    // 消費者，獲取消息
    CH.consume(q.queue, (msg) => {
      var _msg = msg.content.toString()
      var MSG = JSON.parse(_msg)
      log.info(_msg, MSG)
    }, {noAck: true})
  })
  .catch((err) => {
    console.log(err)
  })
}

動態給隊列綁定或者解綁routing key

function toggleBindQueue (routingKey, bind) {
  return new Promise((resolve, reject) => {
    if (!CH) {
      log.error("channel not established")
      reject(new Error("channel not established"))
      return ""
    }
    // 初始化隊列，如果隊列已經存在，就會直接使用
    CH.assertQueue(`${hostName}`, queueAttr)
    .then((q) => {
      // 如果bind是true，就綁定。否則就解綁
      if (bind) {
        log.info(`bindQueue ${hostName} ${topic}`)
        return CH.bindQueue(q.queue, ex, topic)
      } else {
        return CH.unbindQueue(q.queue, ex, topic)
      }
    })
    .then((res) => {
      resolve()
    })
    .catch((err) => {
      reject(err)
      log.error(err)
    })
  })
}

module.exports = {
  connectRabbitMq,
  toggleBindQueue,
  publishMessage
}

使用方法

加入你的服務端用的是Express, 那么在app.js中可以

...
const {connectRabbitMq} = require("./connect-mq.js")
connectRabbitMq()
...

完整代碼

// onnect-mq.js
const amqp = require("amqplib")
// rabbitMQ地址
const {amqpAddrHost} = require("../config/index.js")

// 交換機名稱
const ex = "amq.topic"

const amqpAddr = `amqp://${amqpAddrHost}`

// 讀取HOSTNAME, 在跑多實例時，例如在k8s中，HOSTNAME可以獲取當前pod的名稱
// 多實例時，寫日志，或者建立連接時，最好帶上pod名稱，如果出現問題，也比較好定位哪個pod出現的問題。
const hostName = process.env.HOSTNAME

// 隊列的屬性設置
// 一般來說，最好設置隊列自動刪除autoDelete，當鏈接斷開時，隊列也會刪除，這樣不會產生非常多的無用隊列
// durable是用來的持久化的，最好也可以設置成不持久化

const queueAttr = {autoDelete: true, durable: false}

// 定義channel的引用，當鏈接建立時，所有方法都可以通過引用CH來獲取channel方法
let CH = null


// 向隊列發送消息的函數
function publishMessage (msg) {
  if (!CH) {
    return ""
  }

  msg = JSON.stringify(msg)
  // 指定交換機ex, routing key, 以及消息的內容
  CH.publish(ex, eventBusTopic, Buffer.from(msg))
}

// 當鏈接rabbitMQ斷開時，要主動去重連
function reconnectRabbitMq () {
  log.info("reconnect_rabbit_mq")
  connectRabbitMq()
}

// 鏈接rabbitMQ的主要函數
function connectRabbitMq () {
  amqp.connect(amqpAddr, {
    // 設置connection_name的屬性，可以在rabbitMQ的控制臺的UI上，看到鏈接是來自哪個實例
    clientProperties: {
      connection_name: hostName
    }
  })
  .then((conn) => {
    log.info("rabbitmq_connect_successd")
    // 一定要加上鏈接的報錯事件處理，否則一旦報error錯，如果不處理這個錯誤，程序就會崩潰
    // error是個特別的事件，務必要處理的
    // 報錯就直接去重連
    conn.on("error", (err) => {
      log.error("connect_error " + err.message, err)
      reconnectRabbitMq()
    })
    // 創建channel
    return conn.createChannel()
  })
  .then((ch) => {
    CH = ch
    // 初始化交換機
    ch.assertExchange(ex, "topic", {durable: true})
    // 初始化一個隊列，隊列名就用hostName, 比較容易從對列名上知道是哪個實例創建的隊列
    return ch.assertQueue(hostName, queueAttr)
  })
  .then((q) => {
    // 可以在隊列初始化完畢就立即綁定routing key, 也可以暫時不綁定，后續動態的綁定
    // CH.bindQueue(q.queue, ex, "some.topic.aaa")
    // 消費者，獲取消息
    CH.consume(q.queue, (msg) => {
      var _msg = msg.content.toString()
      var MSG = JSON.parse(_msg)
      log.info(_msg, MSG)
    }, {noAck: true})
  })
  .catch((err) => {
    console.log(err)
  })
}


// 動態給隊列綁定或者解綁routing key
function toggleBindQueue (routingKey, bind) {
  return new Promise((resolve, reject) => {
    if (!CH) {
      log.error("channel not established")
      reject(new Error("channel not established"))
      return ""
    }
    // 初始化隊列，如果隊列已經存在，就會直接使用
    CH.assertQueue(`${hostName}`, queueAttr)
    .then((q) => {
      // 如果bind是true，就綁定。否則就解綁
      if (bind) {
        log.info(`bindQueue ${hostName} ${topic}`)
        return CH.bindQueue(q.queue, ex, topic)
      } else {
        return CH.unbindQueue(q.queue, ex, topic)
      }
    })
    .then((res) => {
      resolve()
    })
    .catch((err) => {
      reject(err)
      log.error(err)
    })
  })
}

module.exports = {
  connectRabbitMq,
  toggleBindQueue,
  publishMessage
}

云服務器 GPU云服務器 webrtc斷線重連 python斷線重連 webrtc 斷線重連 php 斷線重連

文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。

轉載請注明本文地址：http://specialneedsforspecialkids.com/yun/33079.html

Node.js連接RabbitMQ，斷線重連，動態綁定routing key

摘要：官方提供的教程，是基于回調的。下面將給出基于式的寫法。并且實現動態的隊列綁定初始化配置地址交換機名稱讀取在跑多實例時，例如在中，可以獲取當前的名稱多實例時，寫日志，或者建立連接時，最好帶上名稱，如果出現問題，也比較好定位哪個出現的問題。 RabbitMQ官方提供的教程https://www.rabbitmq.com/tuto...，是基于回調的。下面將給出基于Promise式的寫法。...

frank_fun 2019-07-01 16:52 評論0 收藏0
Node.js連接RabbitMQ，斷線重連，動態綁定routing key

摘要：官方提供的教程，是基于回調的。下面將給出基于式的寫法。并且實現動態的隊列綁定初始化配置地址交換機名稱讀取在跑多實例時，例如在中，可以獲取當前的名稱多實例時，寫日志，或者建立連接時，最好帶上名稱，如果出現問題，也比較好定位哪個出現的問題。 RabbitMQ官方提供的教程https://www.rabbitmq.com/tuto...，是基于回調的。下面將給出基于Promise式的寫法。...

Cheriselalala 2019-08-23 12:33 評論0 收藏0
vue項目前端知識點整理

摘要：在如下幾個屬性，表示當前的真實時間，用于和服務器時間同步，表示創建時間，主要用于分頁，以及重連時的判斷，表示是否斷線重連。初始化連接時，將賦值為當前本地時間，連接成功后，將賦值為服務器返回的當前時間，再設置一個定時器，保持時間與服務器一致。 vue項目前端知識點整理微信授權后還能通過瀏覽器返回鍵回到授權頁在導航守衛中可以在next({})中設置replace: true來重定向到改...

bang590 2019-08-26 13:45 評論0 收藏0
20170917 前端開發周報:JavaScript函數式編程、作用域和閉包

摘要：用函數式編程對進行斷舍離當從業的老司機學會函數式編程時，他扔掉了的特性，也不用面向對象了，最后發現了真愛啊作用域和閉包作用域和閉包在里非常重要。旨在幫助非函數式編程的同學，能快速切入到函數式編程的理念。 1、用函數式編程對JavaScript進行斷舍離當從業20的JavaScript老司機學會函數式編程時，他扔掉了90%的特性，也不用面向對象了，最后發現了真愛啊！！！ https:/...

tomener 2019-08-22 11:09 評論0 收藏0
20170917 前端開發周報:JavaScript函數式編程、作用域和閉包

摘要：用函數式編程對進行斷舍離當從業的老司機學會函數式編程時，他扔掉了的特性，也不用面向對象了，最后發現了真愛啊作用域和閉包作用域和閉包在里非常重要。旨在幫助非函數式編程的同學，能快速切入到函數式編程的理念。 1、用函數式編程對JavaScript進行斷舍離當從業20的JavaScript老司機學會函數式編程時，他扔掉了90%的特性，也不用面向對象了，最后發現了真愛啊！！！ https:/...

cyixlq 2019-08-01 17:37 評論0 收藏0