node.js - 在重新连接时使用 node-amqplib 取消订阅特定队列

Question

问题：远程系统重新连接到多个节点 websocket 服务器，为每个系统创建/使用 RabbitMQ 中的专用队列。如果不存在活动连接，则应自动删除队列。Websocket 连接/断开事件处理程序是异步的，相当繁重，观察到的问题是断开事件处理程序在重新连接后完成，导致系统不一致。

主要问题是 RabbitMQ 队列 - 最初的解决方案是为每个连接创建唯一的队列并在断开连接时删除它们。显得很重。

第二种方法是为每个远程系统保留一个专用队列（任何连接的队列名称相同），问题是 assertQueue 为同一个队列添加了消费者。需要找到在不删除队列本身的情况下删除陈旧队列消费者的方法。

Answer 1

解决方案是存储每个远程系统的消费者列表，并在断开连接事件触发取消功能时使用最旧的消费者标签，然后更新给定远程系统的队列消费者列表。

在远程系统连接事件上

import { Replies } from "amqplib";

// bind callback function for queue-specific messages and store returned consumer description
const result: Replies.Consume = await channel.consume(queueName, this.onSomeMessage.bind(this)); 

// update consumers list for the connected remote system
const consumers: Array<string> | undefined = this.consumers.get(remoteId);
if (consumers === undefined) {
  const consumersList: Array<string> = new Array();
  consumersList.push(result.consumerTag);
  this.consumers.set(remoteId, consumersList);
} else {
  consumers.push(result.consumerTag);
}

在远程系统断开事件上

// remove the oldest consumer in the list and update the list itself
// use cancel method of the amqp channel
const consumers = this.consumers.get(remoteId);
if (consumers === undefined) {  
  // shouldn't happen
  console.error(`consumers list for ${remoteId} is empty`);
} else {
  const consumerTag = consumers[0];
  await this.rxchannel.addSetup(async (channel: ConfirmChannel) => {
    await channel.cancel(consumerTag);
    consumers.shift();
  });
}

代码片段来自某些类的方法实现（如果您想知道“this”）。

版权声明（特别是对于德国同事）：此答案中的代码可以在 Beerware （https://en.wikipedia.org/wiki/Beerware）或 MIT 许可（任何人喜欢的）下使用。