发布消息后如何关闭基于Promise的AMQP连接？

看起来你遇到的核心问题是连接复用不当和资源关闭逻辑不完整，导致调用publish()后无法正确关闭AMQP连接。我们一步步来分析并修复：

问题根源 #

1. 重复创建连接：你的Connector是单例，但每次调用connect()都会执行amqp.connect()创建新连接，旧连接被this.connection覆盖后再也无法访问，导致资源泄漏。
2. 未关闭连接：publish()方法里只关闭了Channel，却没有关闭Connection，而Connection才是和RabbitMQ的底层连接。
3. Promise链式调用缺陷：publish()里的connector.connect()没有返回外层Promise，导致外部无法等待消息发送+关闭操作完成，可能在操作结束前进程就退出了。

修复方案 #

1. 重构Connector：复用连接+显式关闭方法 #

首先修改connector.js，让它支持连接复用，并提供可靠的关闭方法：

const amqp = require('amqplib');
class Connector {
  constructor(RabbitMQUrl) {
    this.rabbitMQUrl = RabbitMQUrl;
    this.connection = null;
    this.channel = null;
  }

  async connect() {
    // 复用已存在的连接和Channel，避免重复创建
    if (this.connection && this.channel) {
      return this.channel;
    }

    try {
      this.connection = await amqp.connect(this.rabbitMQUrl);
      // 监听SIGINT信号，确保进程退出时关闭连接
      process.once('SIGINT', async () => {
        await this.close();
      });
      this.channel = await this.connection.createChannel();
      return this.channel;
    } catch (err) {
      console.error('Connection error:', err);
      throw err; // 抛出错误让调用者处理
    }
  }

  // 统一关闭资源：先关Channel，再关Connection
  async close() {
    if (this.channel) {
      await this.channel.close();
      this.channel = null;
    }
    if (this.connection) {
      await this.connection.close();
      this.connection = null;
    }
  }
}

module.exports = new Connector(
  `amqp://${process.env.AMQP_HOST}:5672`
);

2. 重构Publisher：确保消息发送完成后关闭连接 #

修改publisher.js，使用async/await简化逻辑，并确保消息被确认后再关闭连接：

const connector = require('./connector');
class Publisher {
  constructor(exchange, exchangeType) {
    this.exchange = exchange;
    this.exchangeType = exchangeType;
    this.durabilityOptions = {
      durable: true,
      autoDelete: false,
    };
  }

  async publish(msg) {
    try {
      const channel = await connector.connect();
      // 声明交换机
      await channel.assertExchange(
        this.exchange,
        this.exchangeType,
        this.durabilityOptions
      );
      // 启用确认模式，确保消息被RabbitMQ接收
      channel.confirmSelect();
      channel.publish(this.exchange, '', Buffer.from(msg));
      // 等待RabbitMQ确认消息接收
      await channel.waitForConfirms();
      // 关闭连接
      await connector.close();
      console.log('Message published and connection closed successfully');
    } catch (err) {
      console.error('Publish error:', err);
      // 出错时也要尝试关闭连接，避免资源泄漏
      await connector.close().catch(closeErr => console.error('Close error:', closeErr));
    }
  }
}

module.exports = new Publisher(
  process.env.AMQP_EXCHANGE,
  process.env.AMQP_TOPIC
);

关键改进点 #

• 连接复用：避免每次发布消息都创建新连接，减少RabbitMQ的资源消耗。
• 可靠关闭：先关闭Channel再关闭Connection，符合AMQP的资源释放顺序。
• 消息确认：通过confirmSelect()和waitForConfirms()确保消息被RabbitMQ接收后再关闭连接，防止消息丢失。
• 错误处理：在异常分支也尝试关闭连接，避免资源泄漏。

内容的提问来源于stack exchange，提问作者Jarede