MQ优化-永生的消费者

背景

公司有个项目专门做从外部接口取数的，为了保证取数程序的稳定性，我们引入了RabbitMQ，主要用上了MQ的错误重试机制、MQ的分布式的生产消费能力，还有消息处理的可视化能力。基于这些，方便我们安心将任务放在上面跑，并且能水平扩容消费进程数，还可以清晰知道每个队列的任务消费情况。

问题

程序跑了一段时间后发现，MQ消息堆积严重，消息没有消费者消费，消费者无端消失了。
我们这里的每一个 consumer 都绑定了一个 channel，看程序 log 发现，是因为 consumer 的 channel close 掉了，消费依赖于 channel，channel 必须存在，程序才能正常消费。
而 channel close 的原因太多了，有很多不可控的因素。所以我们得思考如何让程序适应 channel 会关闭这一状况，还能正常消费。

解决思路

既然 channel 会 close掉，那我们在 channel 关闭后重新创建 channel 不就行了吗？
于是要考虑怎么让程序识别这个消费者的 channel 状态和如何重新创建 channel。
我们程序是用 amqp 这个包连接 mq 的，而这个包在在创建 channel 时，对外提供了一些事件注册接口，我们可以实现接口，使 channel 处于不同生命周期时会触发事件。例如在 channel close 时触发动作，告诉我这个消费者的 channel closed。

解决方法

我们通过实现 channel close 和 channel error 的接口，在每一个 consumer 实例里设置一个状态量 channelAlive，用来判断当前 consumer 的 channel 是否存活。
当 channel 关闭和发生异常时，将变量 channelAlive 设置为 false，告诉消费者 channel 关闭。
然后设置一个定时程序，每分钟去检测 channelAlive 这个状态，为 false 时重新创建 channel。
具体见代码：

private channelAlive = true

async onModuleInit() {
  const mq = this.mqHelper.getMqConnection()
  const channel = await mq.connection.createChannel()
  await channel.assertQueue(queue)
  await channel.prefetch(preFetch)
  await this.registerChannelEvent(queue, channel)
  this.channelAlive = true
  try {
    await channel.consume(queue, async (msg: any) => {
      // doing something
    })
  } catch (e) {
    this.channelAlive = false
    console.error(`mq consume channel:${queue} error: ${e}`, e)
  }
}

private async registerChannelEvent(queueName: any, channel: any) {
    const that = this
    channel.once('error', function () {
      console.error(queueName, 'channel trigger error')
      that.channelAlive = false
    })
    channel.once('close', function () {
      console.error(queueName, 'channel trigger close')
      that.channelAlive = false
    })
}

/**
   * 定时重连MQ队列
   * 因为程序运行时因为各种因素失去了MQ的连接，重连后他是不会主动去拉取队列中的信息进行消费的，所以需要进行此操作
   */
  async reconnection() {
    schedule.scheduleJob('0 */1 * * * *', async () => {
      if (!this.channelAlive) {
        console.error(this.queueName, 'channel close, reconnecting')
        await this.onModuleInit()
      }
    });
  }

总结

通过以上方法，消费者就永远不会下线，拥有了 "永生" 的消费能力。不过上述实现是基于定时器去重新创建 channel，会有一段时间间隔消费者不可用。如果要求比较高的话，可以改下程序，在 channel close时立即创建 channel 即可。