Java Eclipse Paho 实现 - 自动重新连接

Question

我正在尝试在我的项目中实现 eclipse.paho 以连接 Mqtt Broker(订阅和发布目的)。问题是，当我使用订阅功能(实现 MqttCallback 接口(interface))时，我不知道如果连接丢失我该如何重新连接。 MqttCallback 接口(interface)有一个 connectionLost 方法，但它对于调试导致连接丢失的原因很有用。我搜索但找不到建立自动重新连接的方法。你能建议一个关于这个问题的方法或文件吗？

Answer 1

我正在使用 paho 客户端 1.2.0。使用 MqttClient.setAutomaticReconnect(true) 和接口(interface) MqttCallbackExtended API，感谢 https://github.com/eclipse/paho.mqtt.java/issues/493 ，我可以设法在与代理的连接断开时自动重新连接。

请看下面的代码。

//Use the MqttCallbackExtended to (re-)subscribe when method connectComplete is invoked
public class MyMqttClient implements MqttCallbackExtended {
    private static final Logger logger = LoggerFactory.getLogger(MqttClientTerni.class);
    private final int qos = 0;
    private String topic = "mytopic";
    private MqttClient client;

    public MyMqttClient() throws MqttException {
        String host = "tcp://localhost:1883";
        String clientId = "MQTT-Client";

        MqttConnectOptions conOpt = new MqttConnectOptions();
        conOpt.setCleanSession(true);
        //Pay attention here to automatic reconnect
    conOpt.setAutomaticReconnect(true);
        this.client = new org.eclipse.paho.client.mqttv3.MqttClient(host, clientId);
        this.client.setCallback(this);
        this.client.connect(conOpt);
    }

    /**
     * @see MqttCallback#connectionLost(Throwable)
     */
    public void connectionLost(Throwable cause) {
        logger.error("Connection lost because: " + cause);


    /**
     * @see MqttCallback#deliveryComplete(IMqttDeliveryToken)
     */
    public void deliveryComplete(IMqttDeliveryToken token) {
    }

    /**
     * @see MqttCallback#messageArrived(String, MqttMessage)
     */
    public void messageArrived(String topic, MqttMessage message) throws MqttException {
        logger.info(String.format("[%s] %s", topic, new String(message.getPayload())));
    }

    public static void main(String[] args) throws MqttException, URISyntaxException {
        MyMqttClient s = new MyMqttClient();
    }

    @Override
    public void connectComplete(boolean arg0, String arg1) {
        try {
      //Very important to resubcribe to the topic after the connection was (re-)estabslished. 
      //Otherwise you are reconnected but you don't get any message
        this.client.subscribe(this.topic, qos);
        } catch (MqttException e) {
            e.printStackTrace();
        }

    }
}