NodeJS rhea AMQP 客户端的故障转移

Question

我的机器上有多个 ActiveMQ 运行实例。它们被配置为 shared file system master slave。如果一个 ActiveMQ 服务器宕机，那么另一个应该会自动恢复。这按预期工作。

ActiveMQ第一个实例的相关配置：

<transportConnectors>
    <!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB -->
    <transportConnector name="openwire" uri="tcp://0.0.0.0:61626?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
    <transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
    <transportConnector name="stomp" uri="stomp://0.0.0.0:61623?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
    <transportConnector name="mqtt" uri="mqtt://0.0.0.0:1889?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
    <transportConnector name="ws" uri="ws://0.0.0.0:61625?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
</transportConnectors>

<persistenceAdapter>
    <kahaDB directory="/Users/srikanth.doddi/Downloads/apachemq3.2.4/data/kahadb"/>
</persistenceAdapter>

ActiveMQsecond实例的相关配置：

<transportConnectors>
    <!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB -->
    <transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
    <transportConnector name="amqp" uri="amqp://0.0.0.0:5673?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
    <transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
    <transportConnector name="mqtt" uri="mqtt://0.0.0.0:1883?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
    <transportConnector name="ws" uri="ws://0.0.0.0:61614?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
</transportConnectors>

<persistenceAdapter>
    <kahaDB directory="/Users/srikanth.doddi/Downloads/apachemqold/data/kahadb"/>
</persistenceAdapter>

我在 NodeJS 中使用 AMQP 连接，按以下方式使用 rhea：

var container = require('rhea');
container.on('message', function (context) {
    console.log(context.message.body);
    context.connection.close();
});
container.once('sendable', function (context) {
    context.sender.send({body:'Hello World!'});
});
var connection = container.connect({'port':5672});
connection.open_receiver('examples');
connection.open_sender('examples');

现在，由于我的 ActiveMQ 运行作为主从，如果 5672 出现故障，我希望客户端自动连接到 5673 并继续工作。此检查应连续进行。

在Spring-boot

中是这样实现的

activemq_url=tcp://localhost:61616,tcp://localhost:61626

spring.activemq.broker-url=failover://(${activemq_url})?randomize=false

Answer 1

您必须做一些工作来构建您的客户端代码，以便重新连接发生，rhea 没有 Qpid JMS 或 Artemis Core JMS 客户端等客户端所支持的那种故障转移处理支持。

Rhea 源代码树中有一个重新连接 example，您可以将其用作将该逻辑构建到您自己的应用程序中的起点。

NodeJS rhea AMQP 客户端的故障转移

Failover for NodeJS rhea AMQP client

node.js

activemq

amqp

rhea