amqplib 的 connect() 无自动重连能力，需手动实现重连→重建 channel→重声明交换机/队列→重绑定→重启 consume 的完整流程，并注意防抖、生命周期对齐与错误规避。
amqplib 的 connect() 方法本身不带自动重连能力，断线后连接对象失效，原有 channel、队列声明、绑定关系全部丢失。要实现“断线后自动重新绑定监听”，关键不是让 connect() 自己重绑，而是重建整个连接生命周期：重连 → 重建 channel → 重新声明交换机/队列 → 重新绑定 routing key → 重新 consume。
下面分几个核心环节说明怎么做才稳妥：
连接层必须监听并主动触发重连
connection.on('error') 和 connection.on('close') 是两个必须监听的事件。仅靠 catch 初始化失败不够——因为网络闪断、心跳超时等场景下，连接可能“半死”，error 或 close 才是真实断开信号。
- 不要在这些事件回调里直接调
amqp.connect()，否则容易并发多次重连 - 应引入防抖状态（如
isConnecting = false），只在未连接且未进行中时启动重试
let isConnecting = false;
function connectWithRetry(url, delay = 1000) {
if (isConnecting) return;
isConnecting = true;
amqp.connect(url, { clientProperties: { connection_name: process.env.HOSTNAME } })
.then(conn => {
conn.on('error', err => {
console.error('Connection error:', err.message);
isConnecting = false;
setTimeout(() => connectWithRetry(url, Math.min(delay * 2, 30000)), delay);
});
conn.on('close', () => {
console.warn('Connection closed');
isConnecting = false;
setTimeout(() => connectWithRetry(url, Math.min(delay * 2, 30000)), delay);
});
return setupChannel(conn); // 进入 channel 初始化流程
})
.catch(err => {
console.error('Initial connect failed:', err.message);
isConnecting = false;
setTimeout(() => connectWithRetry(url, Math.min(delay * 2, 30000)), delay);
});
}
Channel 和队列绑定需在每次重连后完整重建
断线后旧 channel 不可用，所有 assertExchange、assertQueue、bindQueue 都得重做。不能复用旧引用，也不能跳过某步（比如只重 bind 不重 assertQueue）。
- 推荐为每个消费者使用独立队列名（如
${HOSTNAME}-${process.pid}），避免多实例冲突 -
autoDelete: true+durable: false可减少残留队列，也降低重连后绑定失败概率
function setupChannel(conn) {
return conn.createChannel().then(ch => {
ch.on('error', console.error); // channel 级错误也要捕获
return ch.assertExchange('amq.topic', 'topic', { durable: true })
.then(() => ch.assertQueue(`worker-${process.env.HOSTNAME}`, { autoDelete: true, durable: false }))
.then(qok => {
// 动态绑定多个 routing key（例如根据配置或服务发现）
['order.created', 'user.updated'].forEach(key => {
ch.bindQueue(qok.queue, 'amq.topic', key);
});
return ch.consume(qok.queue, handleMessage, { noAck: false });
});
});
}
消费逻辑需与 channel 生命周期对齐
ch.consume() 返回的 consumerTag 在 channel 关闭后失效。重连后必须用新 channel 重新 consume，且要确保：
- 旧 consumer 已通过
ch.cancel(consumerTag)显式取消（如果还活着） -
handleMessage函数内部不做跨 channel 引用（如缓存旧 channel 实例） - 若需处理 unack 消息，应在重连前暂停消费，并在恢复后手动
nack(requeue=true)或检查死信
避免常见陷阱
- ❌ 不要全局缓存
CH变量并在断线后直接调CH.publish()—— 它已无效 - ❌ 不要在
connection.on('error')里立即conn.close()—— 此时 conn 可能正处在关闭过程中，会报错 - ❌ 不要省略
clientProperties.connection_name—— 多实例时无法区分哪个 pod 断连，排查困难 - ✅ 建议加健康检查：连接建立后发一条测试消息并等待 ack，确认整条链路可用
不复杂但容易忽略
