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

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

下面將給出基于Promise式的寫法。并且實(shí)現(xiàn)動態(tài)的隊(duì)列綁定

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

// 交換機(jī)名稱
const ex = "amq.topic"

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

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

// 隊(duì)列的屬性設(shè)置
// 一般來說，最好設(shè)置隊(duì)列自動刪除autoDelete，當(dāng)鏈接斷開時，隊(duì)列也會刪除，這樣不會產(chǎn)生非常多的無用隊(duì)列
// durable是用來的持久化的，最好也可以設(shè)置成不持久化

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

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

// 向隊(duì)列發(fā)送消息的函數(shù)
function publishMessage (msg) {
  if (!CH) {
    return ""
  }

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

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

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

function toggleBindQueue (routingKey, bind) {
  return new Promise((resolve, reject) => {
    if (!CH) {
      log.error("channel not established")
      reject(new Error("channel not established"))
      return ""
    }
    // 初始化隊(duì)列，如果隊(duì)列已經(jīng)存在，就會直接使用
    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
}

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

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

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

// 交換機(jī)名稱
const ex = "amq.topic"

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

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

// 隊(duì)列的屬性設(shè)置
// 一般來說，最好設(shè)置隊(duì)列自動刪除autoDelete，當(dāng)鏈接斷開時，隊(duì)列也會刪除，這樣不會產(chǎn)生非常多的無用隊(duì)列
// durable是用來的持久化的，最好也可以設(shè)置成不持久化

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

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


// 向隊(duì)列發(fā)送消息的函數(shù)
function publishMessage (msg) {
  if (!CH) {
    return ""
  }

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

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

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


// 動態(tài)給隊(duì)列綁定或者解綁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 ""
    }
    // 初始化隊(duì)列，如果隊(duì)列已經(jīng)存在，就會直接使用
    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
}