Node.js 中用 amqplib 控制 RabbitMQ 并发的核心是通过 channel.prefetch() 设置 prefetchCount 限制未确认消息数，必须配合 noAck: false 手动应答，且可按 consumer 级别（global=false）独立限流，并需结合 nack、死信队列和错误监听保障稳定性。
在 Node.js 中用 amqplib 控制 RabbitMQ 异步消费的并发，核心是通过 QoS（Quality of Service）机制 限制消费者未确认消息的数量，避免消息堆积压垮服务。这不是靠多开几个 consumer 进程实现的“伪并发”，而是真正可控、可预测的消息分发节奏。
设置 prefetchCount 限制未确认消息数
调用 channel.prefetch() 是最关键的一步。它告诉 RabbitMQ：这个 channel（或这个 consumer）最多同时处理 N 条未 ack 的消息，超出就暂停派发。
- 设为
1：最保守策略。每处理完一条、发一次 ack，才给下一条。适合任务耗时长、资源占用高或状态强依赖的场景 - 设为
5或10：平衡吞吐与稳定性。常见于中等复杂度任务，比如发通知、写日志、轻量计算 - 不设置或设为
0：RabbitMQ 默认尽可能快地推送所有消息，极易导致 consumer 内存溢出或超时失败
必须配合手动应答（noAck: false）
prefetch 只在手动应答模式下生效。如果 noAck: true，RabbitMQ 发出去就认为成功，完全不关心你有没有处理完，限流也就失去意义。
正确写法示例：
channel.consume(queueName, (msg) => {
const data = JSON.parse(msg.content.toString());
// 执行业务逻辑（如数据库更新、API 调用）
processTask(data).then(() => {
channel.ack(msg); // 处理成功才 ack
}).catch(err => {
channel.nack(msg, false, true); // 失败可选择重回队列重试
});
}, { noAck: false });
按 consumer 级别而非 channel 级别控制更灵活
amqplib 的 prefetch(count, global) 第二个参数 global 默认为 false，表示该限制作用于当前 consumer 实例。这意味着：
- 一个 channel 启动多个 consumer（例如多个
channel.consume），每个都独立受自己设置的 prefetch 约束 - 适合做细粒度分流，比如按任务类型启动不同 consumer，各自配置不同并发数
- 若设
global: true，则整个 channel 共享这个上限，适用性较窄
结合错误处理与重试避免卡死
仅靠 prefetch 不足以应对异常。要防止某条消息反复失败导致后续消息一直等不到机会：
- 对失败消息使用
channel.nack(msg, false, false)拒绝并丢弃，或true让其重回队尾（注意避免无限循环） - 更稳妥的做法是将失败消息转发到专门的死信队列（DLX），单独监控和人工干预
- 在 consumer 启动时监听
channel.on('error')，确保连接异常后能重建 channel 并重新设置 prefetch
