Mqtt（paho）重连机制

本文是使用Java语言，eclipse paho的实现方式，去调用MQTT服务器端，编写的MqttClient代码中针对MQTT服务器重启定制重连机制所遇到的问题进行汇总。

1.1编写MqttConnection类，创建MQTT连接

public synchronized boolean connect() {
        try {
            if(null == client) {
                //host为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，
                // MemoryPersistence设置clientid的保存形式，默认为以内存保存
                client = new MqttClient(host, client_id, new MemoryPersistence());
                //设置回调
                client.setCallback(new PushCallBack(MqttConnection.this, devDao));
            }
            //获取连接配置
            getOption();
            client.connect(option);
            log.info("[MQTT] connect to Mqtt Server success...");
            return isConnected();
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

private void getOption() {
        //MQTT连接设置
        option = new MqttConnectOptions();
        //设置是否清空session,false表示服务器会保留客户端的连接记录，true表示每次连接到服务器都以新的身份连接
        option.setCleanSession(true);
        //设置连接的用户名
        option.setUserName(userName);
        //设置连接的密码
        option.setPassword(password.toCharArray());
        //设置超时时间 单位为秒
        option.setConnectionTimeout(outTime);
        //设置会话心跳时间 单位为秒 服务器会每隔(1.5*keepTime)秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
        option.setKeepAliveInterval(keepTime);
        option.setAutomaticReconnect(true);
        //setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息
//            option.setWill(topic, "close".getBytes(), 2, true);
    }

1.2编写PushCallBack回调类，实现重连

public void connectionLost(Throwable throwable) {
        // 连接丢失后，一般在这里面进行重连
        log.info("[MQTT] 连接断开，30S之后尝试重连...");
        while(true) {
            try {
                Thread.sleep(30000);
                mqttConn.reConnect();
                break;
            } catch (Exception e) {
                e.printStackTrace();
                continue;
            }
        }
    }

异常出现：

若在connectionLost()方法中直接循环调用MqttConnection类中connect()方法，

去实现重连机制的话，会出现在第一次重连成功后，一直断开连接再重连再断开连接再重连的死循环中。

异常定位：

原因在connect()方法中的这句：

 

重新new上一个client_id相同的MqttClient，client_id是MQTT client的唯一标识，client_id不能重复。

这样就会出现重连时创建的MqttClient，使程序中初始化时创建的MqttClient断开连接，断开连接后就会回滚到connectionLost方法中，

然后此方法中又会继续重连，就出现上述的断开连接再重连再断开连接再重连的死循环。

异常解决：

在MqttConnection连接类中定义一个重连的方法：

//断线重连
    public void reConnect() throws Exception {
        if(null != client) {
            client.connect(option);
        }
    }

不需要重新new一个MqttClient，只需要调用connect方法就OK了，eclipse paho就会把之前的连接重新创建起来。

另外分享setRetained()方法：

/**
     * 发布消息
     * @param topic     消息主题
     * @param qos       消息传输质量
     * @param message   消息内容
     */
    public void publish(String topic, int qos, String message) throws Exception {
        MqttTopic mqttTopic = client.getTopic(topic);
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setQos(qos);
        //是否设置保留消息，若为true，后来的订阅者订阅该主题时仍可接收到该消息
        mqttMessage.setRetained(false);
        mqttMessage.setPayload(message.getBytes());
        MqttDeliveryToken token = mqttTopic.publish(mqttMessage);
        token.waitForCompletion();
        log.info("[MQTT] publish message : " + token.isComplete() +
                ",{topic : " + topic + ", message : " + message + "}");
    }

setRetained()：消息保留机制，若设置为true，mqtt服务器会保留每次发布的消息，

若订阅某主题的客户端重启，则会把此主题之前发布的消息重新推送到客户端。