ActiveMQ + MQTT + 订阅“ActiveMQ.Advisory.Connection”

Question

这是上下文：

Java 应用程序通过 MQTT (Paho 0.4.0) 从 ActiveMQ 5.9.1 订阅主题“ActiveMQ.Advisory.Connection”：

public class SupervisorMqttClient implements MqttCallback {

    private MqttClient client = null;
    private MemoryPersistence persistence = null;
    private MqttConnectOptions connOpts = null;

    private final int STATUS_OK = 0;
    private final int STATUS_ERROR = 1;

    private String mqttServer = null;
    private String clientId = null;
    private int status = STATUS_OK;

    public SupervisorMqttClient() {
        try {
            this.init();
        } catch (MqttException e) {
            Logger.error(e.getLocalizedMessage());
            Logger.debug(e);
        }
    }

    private void init() throws MqttException {
        Properties props = PropertiesManager.getInstance("supervisor");

        mqttServer = props.getProperty("supervisor.mqtt.server");
        String supervisorID = props.getProperty("supervisor.mqtt.client.number");
        clientId = Supervisor.APP_NAME+"-"+supervisorID;

        connOpts = new MqttConnectOptions();
        connOpts.setKeepAliveInterval(30);
        connOpts.setCleanSession(true); // important non-durable

        persistence = new MemoryPersistence();

        client = new MqttClient(mqttServer, clientId, persistence);
        connectAndSubscribe();
    }

    private void connectAndSubscribe() throws MqttSecurityException, MqttException {
        try {
            client.connect(connOpts);
            client.setCallback(this);
            client.subscribe("ActiveMQ/Advisory/Connection");
        } catch (MqttSecurityException e) {
            Logger.error(e.getLocalizedMessage());
            Logger.debug(e);
        } catch (MqttException e) {
            Logger.error(e.getLocalizedMessage());
            Logger.debug(e);
            processError(e);
        }
    }

    public void publish(String orderType, JSONObject jsonExtraData) {
        if (status == STATUS_ERROR) {
            connectAndSubscribe();
        }
        if (status == STATUS_OK) {
            // some code here
        }
    }

    @Override
    public void connectionLost(Throwable err) {
        Logger.info("Connection lost");
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken arg0) {
        Logger.info("deliveryComplete");
    }

    @Override
    public void messageArrived(String topic, MqttMessage msg) throws Exception {
        System.out.println("MQTT Mesage Arrived[" + topic + "] Msg[" + msg.toString() + "]");
    }

    private void processError(MqttException e) {
        status = STATUS_ERROR;
        try {
            if (client.isConnected()) {
                Logger.error("disconnecting");
                client.disconnect();
            }
        } catch (MqttException ex) {
            Logger.error(ex.getLocalizedMessage());
            Logger.debug(ex);
        }
    }
}

与 ActiveMQ 的连接建立良好。本主题提供有关 ActiveMQ 中的连接（打开/关闭）的信息，但我的问题是我捕获的消息是空的：

MQTT 消息到达[ActiveMQ/Advisory/Connection] Msg[]

有什么方法可以使用 MQTT 捕获它们？还是我应该为此使用 JMS？

谢谢， 乔恩·安德。

Answer 1

问题是您希望 MQTT 客户端在咨询主题上接收什么作为消息正文。公告通常包含作为消息属性的大部分信息，但是这些信息无法映射到 MQTT，因为 MQTT 消息没有属性。 Connection 建议的正文是用于创建连接的 ConnectionInfo 对象的副本。在 MQTT 方面，您无能为力，因为您将收到的只是该对象的序列化字节，您将无法做任何事情。