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

Java Eclipse Paho Implementation - Auto reconnect

提问by Black Glix

I'm trying to implement eclipse.pahoin my project to connect Mqtt Broker (Both subscribing and publishing purpose). The problem is, when I using the subscribing feature (Implementing MqttCallbackinterface), I couldn't figure our how can I reconnect if the connection lost. MqttCallback interface has a connectionLost method, but it is useful for the debug what causes the connection lost. I searched but couldn't find a way to establish auto reconnect. Can you suggest a way or document about this problem?

我正在尝试eclipse.paho在我的项目中实现连接 Mqtt Broker（订阅和发布目的）。问题是，当我使用订阅功能（实现MqttCallback接口）时，我无法弄清楚如果连接丢失如何重新连接。MqttCallback 接口有一个 connectionLost 方法，但它对于调试导致连接丢失的原因很有用。我搜索但找不到建立自动重新连接的方法。您能否提出有关此问题的方法或文档？

采纳答案by hardillb

The best way to do this is to structure your connection logic so it lives in a method on it's own so it can be called again from the connectionLostcallback in the MqttCallbackinstance.

最好的方法是构建您的连接逻辑，使其独立存在于一个方法中，以便可以从实例中的connectionLost回调中再次调用它MqttCallback。

The connectionLostmethod is passed a Throwable that will be the exception that triggered the disconnect so you can make decisions about the root cause and how this may effect when/how you reconnect.

该connectionLost方法传递了一个 Throwable，它将是触发断开连接的异常，因此您可以决定根本原因以及这可能如何影响您何时/如何重新连接。

The connection method should connect and subscribe to the topics you require.

连接方法应该连接并订阅您需要的主题。

Something like this:

像这样的东西：

public class PubSub {

  MqttClient client;
  String topics[] = ["foo/#", "bar"];
  MqttCallback callback = new MqttCallback() {
    public void connectionLost(Throwable t) {
      this.connect();
    }

    public void messageArrived(String topic, MqttMessage message) throws Exception {
      System.out.println("topic - " + topic + ": " + new String(message.getPayload()));
    }

    public void deliveryComplete(IMqttDeliveryToken token) {
    }
  };

  public static void main(String args[]) {
    PubSub foo = new PubSub();
  }

  public PubSub(){
    this.connect();
  }

  public void connect(){
    client = new MqttClient("mqtt://localhost", "pubsub-1");
    client.setCallback(callback);
    client.connect();
    client.subscribe(topics);
  }

}

回答by Doug

To use auto reconnect, just set setAutomaticReconnect(true)on the MqttConnectOptionsobject.

要使用自动重新连接，只需setAutomaticReconnect(true)在MqttConnectOptions对象上设置。

MqttAndroidClient mqttClient = new MqttAndroidClient(context, mqttUrl, clientId);

MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
mqttConnectOptions.setAutomaticReconnect(true);

mqttClient.connect(mqttConnectOptions, null, mqttActionListener());

回答by florins

I'm using the paho client 1.2.0. With the MqttClient.setAutomaticReconnect(true) and interface MqttCallbackExtended API, and thanks to https://github.com/eclipse/paho.mqtt.java/issues/493, I could manage to reconnect automatically when the connection to broker is down.

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

See below the code.

请参阅下面的代码。

//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();
        }

    }
}