9

クライアントがRabbitMQキューからメッセージを消費するsocket.ioサーバーを実行するnode.jsスクリプトがあります。最近、Amazon AWS に移行し、RabbitMQ は 2 台のマシン (冗長インスタンス) のクラスターになりました。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());
});
4

1 に答える 1