分布式MQTT消息订阅-发布框架：高可用性ActiveMQ

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.pool.PooledConnectionFactory;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
 
public class MqttActiveMQSubPub {
 
    private static final String BROKER_URL = "tcp://localhost:61613";
    private static final String CLIENT_ID = "JavaClient";
    private static final String TOPIC = "MQTT_Examples_Topic";
 
    public static void main(String[] args) {
        // 创建MQTT客户端，使用PooledConnectionFactory提高连接复用效率
        PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory(new ActiveMQConnectionFactory(BROKER_URL));
        MqttClient client = null;
        try {
            client = new MqttClient(pooledConnectionFactory.getClientURI(), CLIENT_ID, new MemoryPersistence());
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setCleanSession(true);
            // 设置连接认证信息，如果ActiveMQ需要
            // connOpts.setUserName("username");
            // connOpts.setPassword("password".toCharArray());
 
            // 连接到MQTT代理
            client.connect(connOpts);
 
            // 订阅主题
            client.subscribe(TOPIC);
 
            // 回调实现，用于处理消息接收
            client.setCallback(new MqttCallback() {
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    System.out.println("Received message: " + new String(message.getPayload()));
                }
 
                public void connectionLost(Throwable cause) {
                    System.out.println("Connection lost");
                }
 
                public void deliveryComplete(IMqttDeliveryToken token) {
                    System.out.println("Delivery complete");
                }
            });
 
            // 发布消息
            MqttMessage message = new MqttMessage("Hello MQTT".getBytes());
            client.publish(TOPIC, message);
 
        } catch (MqttException me) {
            System.out.println("reason " + me.getReasonCode());
            System.out.println("msg " + me.getMess