ActiveMQ HA 故障转移

Question

ActiveMQ 集群（主从）是否在故障转移模式下提供高可用性？ 我正在发布有关某个主题的消息，但是当我杀死活动节点并且我的消费者连接到另一个节点时，它会丢失几条消息。 我只需要主题模式，因为我需要有不同的 Pod 同时消费消息。我们已经对持久消息和订阅进行了测试，但我们遇到了同样的问题。

如果 ActiveMQ 不支持此功能，是否有其他代理可以做到这一点？

我们已经使用 Spring boot 来开发发送者和消费者应用程序。

发件人：

为了发送消息，我们使用了向主题发送 5000 条消息的 API REST。如果某些 convertAndSend 失败（ActiveMQ 节点 1 被杀死），我们会重试它，直到故障转移。

这是我的发送者类：

@RestController
public class RestApiController {

    @Autowired
    private JmsTemplate jmsTemplate;

    @RequestMapping(value = "/produce")
    public String produce() {
        String result = "Done";
        for (int i = 0; i < 5000; i++) {
            boolean repetir = true;
            while (repetir) {
                try {
                    jmsTemplate.convertAndSend("TestTopic", "Message " + i);
                    repetir = false;
                    result = "Done";
                } catch (JmsException e) {
                    e.printStackTrace();
                    result = "ERROR";
                }
            }
        }
        return result;
    }
}

这是发送者的application.properties：

spring.activemq.broker-url=failover:(tcp://172.18.13.45:61616,tcp://172.18.13.45:61626)?initialReconnectDelay=1&backup=true spring.activemq.user=admin
spring.activemq.password=admin 
spring.jms.pub-sub-domain=true 
server.port=8081

消费者：

这是我的监听类：

@Service
public class ContactTransactionReceiver {

    @JmsListener(destination = "TestTopic")
    public void receiveMessageSendMessage(Message message) throws Exception {
        System.out.println(((TextMessage) message).getText());
    }
}

这是消费者的application.properties：

spring.activemq.broker-url=failover:(tcp://172.18.13.45:61616,tcp://172.18.13.45:61626)?initialReconnectDelay=1&backup=true
spring.activemq.user=admin
spring.activemq.password=admin
spring.jms.pub-sub-domain=true
server.port=8082

ACTIVEMQ 节点 1

我们在activemq.xml中包含了这个配置，用于HA，指的是node2：

    <persistenceAdapter>
        <kahaDB directory="${activemq.data}/kahadb"/>
    </persistenceAdapter>
    <networkConnectors>
        <networkConnector uri="static:(tcp://172.18.13.45:61616,tcp://172.18.13.45:61626)" />
    </networkConnectors>

我们已经证明太主从了：

    <persistenceAdapter>
        <kahaDB directory="${activemq.data}/kahadb"/>
    </persistenceAdapter>
    <networkConnectors>
        <networkConnector uri="masterslave:(tcp://172.18.13.45:61616,tcp://172.18.13.45:61626)" />
    </networkConnectors>

ACTIVEMQ 节点 2

我们在activemq.xml中包含了这个配置，用于HA，指的是node2：

    <persistenceAdapter>
        <kahaDB directory="${activemq.data}/kahadb"/>
    </persistenceAdapter>
    <networkConnectors>
        <networkConnector uri="static:(tcp://172.18.13.45:61626,tcp://172.18.13.45:61616)" />
    </networkConnectors>

我们已经证明太主从了：

    <persistenceAdapter>
        <kahaDB directory="${activemq.data}/kahadb"/>
    </persistenceAdapter>
    <networkConnectors>
        <networkConnector uri="masterslave:(tcp://172.18.13.45:61626,tcp://172.18.13.45:61616)" />
    </networkConnectors>

您可以在以下位置找到完整的代码和 ActiveMQ 配置文件：

https://github.com/PedroRamirezTOR/ActiveMQ-HA-Sender-Consumer.git

提前致谢！

Answer 1

我注意到在您的 xml 中，主从都指的是本地 KashaDB。我相信要实现真正的 HA 故障转移，您需要引用共享网络驱动器。

http://activemq.apache.org/shared-file-system-master-slave

<persistenceAdapter>
  <kahaDB directory="/sharedFileSystem/sharedBrokerData"/>
</persistenceAdapter>

Answer 2

ActiveMQ 5.x 确实通过主从配置支持高可用性。详情请见the documentation。