我们有一个运行 socket.io 服务器的 node.js 脚本，它的客户端使用来自 RabbitMQ 队列的消息。我们最近迁移到了 Amazon AWS，而 RabbitMQ 现在是一个由两台机器（冗余实例）组成的集群。AMQP 连接有时会丢失（这是来自具有冗余 VM 的高可用性环境的限制，我们必须应对它），如果尝试重新连接，DNS 会选择连接到哪个实例（它是一个具有数据复制功能的集群，因此连接到哪个实例并不重要）。
问题是从未尝试重新连接；一段时间后，当连接丢失时，amqp.node 显然没有注意到连接已丢失。此外，消费者停止接收消息，socket.io 服务器只是停止接受新连接。
我们在 RabbitMQ URL 处设置了 55 秒的心跳超时（不要与 socket.io 心跳超时混淆），并正在使用 amqp.node 的回调 API 检查“错误”和“关闭”事件，但它们显然从未发出过。队列期望消费的消息被确认。我们希望节点脚本检测到丢失的连接并自行完成，因此环境将自动启动一个新进程并再次建立连接。
这是代码，也许我们在 amqp.node 回调 API 或其他方面做错了。
var express = require('express');
app = express();
var http = require('http');
var serverio = http.createServer(app);
var io = require('socket.io').listen(serverio, { log: false });
var socket;
var allcli = [];
var red, blue, green, magenta, reset;
red = '\033[31m';
blue = '\033[34m';
green = '\033[32m';
magenta = '\033[35m';
orange = '\033[43m';
reset = '\033[0m';
var queue = 'ha.atualizacao_mobile';
var urlRabbit = 'amqp://login:password@host?heartbeat=55' // Amazon
var amqp = require('amqplib/callback_api');
var debug = true;
console.log("Original Socket.IO heartbeat interval: " + io.get('heartbeat interval') + " seconds.");
io.set('heartbeat interval', 10 * 60);
console.log("Hearbeat interval changed to " + io.get('heartbeat interval') + " seconds to reduce battery consumption in the mobile clients.");
console.log("Original Socket.IO heartbeat timeout: " + io.get('heartbeat timeout') + " seconds.");
io.set('heartbeat timeout', 11 * 60);
console.log("Heartbeat timeout set to " + io.get('heartbeat timeout') + " seconds.");
io.sockets.on('connection', function(socket){
socket.on('error', function (exc) {
console.log(orange+"Ignoring exception: " + exc + reset);
});
socket.on('send-indice', function (data) {
// Some business logic
});
socket.on('disconnect', function () {
// Some business logic
});
});
function updatecli(data){
// Some business logic
}
amqp.connect(urlRabbit, null, function(err, conn) {
if (err !== null) {
return console.log("Error creating connection: " + err);
}
conn.on('error', function(err) {
console.log("Generated event 'error': " + err);
});
conn.on('close', function() {
console.log("Connection closed.");
process.exit();
});
processRabbitConnection(conn, function() {
conn.close();
});
});
function processRabbitConnection(conn, finalize) {
conn.createChannel(function(err, channel) {
if (err != null) {
console.log("Error creating channel: " + err);
return finalize();
}
channel.assertQueue(queue, null, function(err, ok) {
if (err !== null) {
console.log("Error asserting queue " + queue + ": " + err);
return finalize();
}
channel.consume(queue, function (msg) {
if (msg !== null) {
try {
var dataObj = JSON.parse(msg.content);
if (debug == true) {
//console.log(dataObj);
}
updatecli(dataObj);
} catch(err) {
console.log("Error in JSON: " + err);
}
channel.ack(msg);
}
}, null, function(err, ok) {
if (err !== null) {
console.log("Error consuming message: " + err);
return finalize();
}
});
});
});
}
serverio.listen(9128, function () {
console.log('Server: Socket IO Online - Port: 9128 - ' + new Date());
});