Mqtt Paho 客户端连接始终打开

Question

我已经创建了 mqtt 连接 suscribeData() 如下，以便从多个设备 24/7 接收数据

@RequestMapping("/suscribe")
@ResponseBody
public String suscribeData(){

    connOpt= new MqttConnectOptions();
    connOptPublish= new MqttConnectOptions();
    boolean result=false;
    int retry=0;
    while(!result){
        try{
            result=initializeMqTTParameters(retry); 
            retry++;
            if(!result){
                Thread.sleep(1000);
            }

        }catch(Exception ex){

        }
        if(retry>10){
            return "mqtterror";
        }

    }
        suscribeReceivedData("DATA/#",2);
        Calendar cal=TimeZoneJu.getCalenderInstance();
        ModemOnlineStatus status=new ModemOnlineStatus();
        status.setActiondate(cal.getTime());
        status.setActiontime(cal.getTime());
        status.setModemid("mqttreceive");
        status.setRemark("online");
        status.setStatus(true);
        try{

            service.checkAndInsertModemStatus(true, status);    
        }catch(Exception ex1){

        }
        return "ok";
}

checkAndInsertModemStatus()函数用于保存连接的在线离线状态我用cronjobs检查另一个url的连接。

以下是我的 initializeMqTTParameters

public boolean initializeMqTTParameters(int retry){
        String clientID = M2MIO_THING;
        connOpt = new MqttConnectOptions();

        connOpt.setCleanSession(false);
        connOpt.setUserName(M2MIO_USERNAME);
        connOpt.setPassword(M2MIO_PASSWORD_MD5.toCharArray());
        connOpt.setWill("WILL/mqttreceive", "mqttreceive:disconnected".getBytes(), 2, true);
        try {

            if(retry%2==0){
                MQTTReceive.myClient = new MqttClient(BROKER_URL, "mqtt_virtual_received_sus_1");   
            }else{
                MQTTReceive.myClient = new MqttClient(BROKER_URL, "mqtt_virtual_received_sus_2");   
            }

            MQTTReceive.myClient.setCallback(this);
            MQTTReceive.myClient.connect(connOpt);

        } catch (MqttException e) {
            System.out.println(e.getMessage());
            return false;
        }
        return true;
    }

关于 connectionLost 我正在尝试更新数据库中的状态并尝试重新连接

@Override
    public void connectionLost(Throwable ex) {
        System.out.println("Mqtt Received Connection Lost"+ex.getMessage()); 
        Calendar cal=TimeZoneJu.getCalenderInstance();
        ModemOnlineStatus status=new ModemOnlineStatus();
        status.setActiondate(cal.getTime());
        status.setActiontime(cal.getTime());
        status.setModemid("mqttreceive");
        status.setRemark("offline");
        status.setStatus(false);
        try{

        service.updateModemStatus(false, status);   
    }catch(Exception ex1){

    }

    suscribeData();
}

从上面的代码我得到我的连接是在线的，但我停止接收订阅主题的数据，所以我尝试了另一种方法，我从数据库的另一个函数检查连接状态，就像我在连接和离线时设置连接在线connectionLost 并且我可以调用 connect(suscribeData()) 但它在几天内让我在 tomcat 中打开了太多打开的文件我如何确保以前打开的文件应该关闭

like System.exit(1) on connectionLost() 这样当前进程将关闭，我将在检查状态后重新连接

或者如果我可以使用任何其他逻辑来维护我的连接，请建议我

Answer 1

你的 catch 块在方法 connectionLost 中什么都不做。它应该做这样的事情：

 catch (Exception e)
         {
              e.printStackTrace();
              System.exit(1);
          }

您还应该在不需要时断开您的 MQTT 客户端。即处理结束：

myClient.disconnect();