生产环境中Node.js使用RabbitMQ出现内存泄漏问题

核心问题定位 #

你的代码仅在发送1MB以上大消息时出现内存泄漏，小消息无此问题，说明泄漏根源与大消息的内存生命周期管理相关：

• amqplib的sendToQueue默认是无确认异步发送，大消息会在底层网络缓冲区堆积，未被及时释放
• 手动调用GC属于临时 workaround，无法从根本上解决内存持有问题

可行解决方案 #

方案1：开启Confirm模式等待消息ACK

强制等待RabbitMQ确认消息接收后再释放资源，确保amqplib不会长期持有大消息的Buffer引用：

import { Connection, Channel, connect } from 'amqplib';
import * as msgpack from '@msgpack/msgpack';

async emit(eventName: string, data: any, route: string = '*'): Promise<boolean> {
  let serializedMessage: Uint8Array | null = null;
  let channel: Channel | undefined;

  try {
    const messageData = { eventName, data };
    serializedMessage = msgpack.encode(messageData);

    channel = await this.getChannel();
    if (!channel) throw new Error('Channel is not available');

    // 仅在通道初始化时调用一次即可，无需每次发送重复执行
    if (!channel._confirming) await channel.confirmSelect();

    const buffer = Buffer.from(serializedMessage);
    // 等待RabbitMQ确认消息已接收
    await new Promise<void>((resolve, reject) => {
      channel!.sendToQueue(route, buffer, (err) => err ? reject(err) : resolve());
    });

    log.debug(`Memory usage after sending message: ${JSON.stringify(process.memoryUsage())}`);
    return true;
  } catch (error) {
    log.error('Failed to emit message', error);
    throw error;
  } finally {
    serializedMessage = null;
    // 无需手动清空Buffer，ACK后GC会自动回收
  }
}

方案2：避免Buffer内存复制

msgpack.encode返回的Uint8Array可直接传递给sendToQueue（通过类型断言绕过类型检查），减少一次大内存复制操作：

// 替换原Buffer.from逻辑
const result = await new Promise<void>((resolve, reject) => {
  channel!.sendToQueue(route, serializedMessage as unknown as Buffer, (err) => err ? reject(err) : resolve());
});

方案3：限制通道未确认消息数量

通过prefetch控制通道上未确认消息的最大数量，避免大消息无限堆积：

// 在通道初始化时设置（例如限制为10条）
await channel.prefetch(10);

当未确认消息达到阈值时，sendToQueue会阻塞直到有消息被确认，从源头控制内存占用。

方案4：检查通道复用逻辑

确保getChannel()返回的通道状态正常：

• 通道出现错误时及时关闭并重新创建
• 避免创建过多通道，保持合理的复用数量
• 定期清理无效/闲置通道

优先级推荐 #

优先尝试方案1，这是解决大消息内存泄漏最直接的方式，从根本上确保消息发送完成后资源被释放；搭配方案3可进一步控制内存峰值。

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