增加查询单个设备（RTU或FBox）是否在线的功能接口

liurunyu

2025-07-31 85f4fbf75505c35a4d0e53515ffb3affb6fd95f2

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/server/mqtt/MqttPubMsgNode.java

@@ -3,6 +3,7 @@
import com.dy.common.mw.protocol4Mqtt.MqttPubMsg;
import com.dy.common.queue.NodeObj;
import com.dy.rtuMw.server.ServerProperties;
import com.dy.rtuMw.server.forTcp.RtuLogDealer;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.eclipse.paho.client.mqttv3.MqttClient;
@@ -19,12 +20,16 @@
    public MqttPubMsg result ;//下行命令
    public Long cachTime ;//缓存时刻
    public boolean onceReceivedResult ;//已经收到命令应答
    public long lastSendTime = 0 ;//上次发送时间
    public int sendTimes = 0 ;//发送次数


    public MqttPubMsgNode(MqttPubMsg result){
        this.result = result ;
        this.cachTime = System.currentTimeMillis() ;
        this.onceReceivedResult = false ;
        this.lastSendTime = 0L ;
        this.sendTimes = 0 ;
    }

    /**
@@ -35,9 +40,13 @@
    public boolean dealSelf(Long now){
        if(this.onceReceivedResult){
            //已经收到命令结果
            //记录状态
            //RtuStatusDealer.commandSuccess(this.result.rtuAddr, this.result.downCode, this.result.downCodeName);
            return true ;
        }
        if(this.sendTimes >= (1 + MqttUnit.confVo.reSendTimesByNoResult)){
            return this.decideRemoveNodeFromCach(now, null) ;
        }
        if(this.lastSendTime != 0 && now - this.lastSendTime >= MqttUnit.confVo.sendInterval){
            return this.decideRemoveNodeFromCach(now, null) ;
        }
        boolean noConnect2MqSv = false ;
        MqttManager mqttManager = MqttManager.getInstance() ;
@@ -46,7 +55,7 @@
        noConnect2MqSv = mqttManager.poolIsClose() ;
        if(noConnect2MqSv){
            //未曾连接MQTT服务器
            return this.decideRemoveNodeFromCach(now) ;
            return this.decideRemoveNodeFromCach(now, null) ;
        }else{
            try {
                //如果网络不好或断网，此处用时较长
@@ -60,32 +69,40 @@
        }
        if(noConnect2MqSv){
            //未曾连接MQTT服务器
            return this.decideRemoveNodeFromCach(now) ;
            return this.decideRemoveNodeFromCach(now, null) ;
        }else{
            if(mqttClient != null && mqttClient.isConnected()){
                try {
                    mqttManager.publishMsg(mqttClient, this.result.topic, this.result.msg);
                    log.info("发布MQTT消息（主题=" + this.result.topic + "）" + this.result.msg);
                    mqttManager.publishMsg(mqttClient, this.result.topic.longName(), this.result.msg);
                    this.sendTimes ++ ;
                    this.lastSendTime = System.currentTimeMillis() ;
                    DevStatusDealer.afterSendPubMessage(this.result.deviceId);
                    RtuLogDealer.log4Mqtt(this.result.deviceId, "发布消息    主题：" + this.result.topic.longName() + "   消息：" + this.result.msg);
                    log.info("发布MQTT消息（主题=" + this.result.topic.longName() + "）" + this.result.msg);
                }catch (Exception e){
                    log.error("MQTT发布消息失败（主题=" + this.result.topic + "）" , e);
                    log.error("MQTT发布消息失败（主题=" + this.result.topic.longName() + "）" , e);
                }finally {
                    mqttManager.pushMqttClient(mqttClient);
                }
                return false ;
                if(this.result.hasResponse){
                    return false ;
                }else{
                    return true ;
                }
            }else{
                //未曾连接MQTT服务器
                return this.decideRemoveNodeFromCach(now) ;
                return this.decideRemoveNodeFromCach(now, false) ;
            }
        }
    }

    private boolean decideRemoveNodeFromCach(Long now){
        if(!this.result.isCacheForOffLine){
    private boolean decideRemoveNodeFromCach(Long now, Boolean isOffLine){
        if(isOffLine != null && isOffLine.booleanValue() && !this.result.isCacheForOffLine){
            //不在线命令不缓存
            return true ;
        }else{
            //不在线命令缓存
            if(now - this.cachTime >= ServerProperties.offLineCacheTimeout){
            if(now - this.cachTime >= MqttUnit.confVo.comCacheTimeout){
                //缓存超时
                return true ;
            }