Merge branch 'master' of http://8.140.179.55:20000/r/pipIrr-SV

zhubaomin

2 天以前 70f61b0638e32274d0e5f9b972ce2a18f139f1b4

Merge branch 'master' of http://8.140.179.55:20000/r/pipIrr-SV

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/channel/mqtt/MqttClientPool.java

@@ -13,8 +13,8 @@

    private final GenericObjectPool<MqttClient> pool;

    public MqttClientPool(String broker, String username, String password, int maxConnections) {
        MqttClientPooledObjectFactory factory = new MqttClientPooledObjectFactory(broker, username, password);
    public MqttClientPool(String broker, String username, String password, int maxConnections, boolean useMemoryPersistence) {
        MqttClientPooledObjectFactory factory = new MqttClientPooledObjectFactory(broker, username, password, useMemoryPersistence);
        GenericObjectPoolConfig<MqttClient> config = new GenericObjectPoolConfig<>();
        config.setMaxTotal(maxConnections);
        config.setMaxIdle(maxConnections);

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/channel/mqtt/MqttClientPooledObjectFactory.java

@@ -5,6 +5,7 @@
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

/**
 * @Author: liurunyu
@@ -16,18 +17,26 @@
    private final String broker;
    private final String username;
    private final String password;
    private final Boolean useMemoryPersistence;

    public MqttClientPooledObjectFactory(String broker, String username, String password) {
    public MqttClientPooledObjectFactory(String broker, String username, String password, boolean useMemoryPersistence) {
        this.broker = broker;
        this.username = username;
        this.password = password;
        this.useMemoryPersistence = useMemoryPersistence;
    }

    @Override
    public MqttClient create() throws Exception {
        String clientId = MqttClient.generateClientId();
        MqttClient client = new MqttClient(broker, clientId);

        MqttClient client = null ;
        // ä½¿ç”¨å†…存持久化而非默认的文件持久化
        if (useMemoryPersistence) {
            MemoryPersistence persistence = new MemoryPersistence();
            client = new MqttClient(broker, clientId, persistence);
        }else{
            client = new MqttClient(broker, clientId);
        }
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(username);
        options.setPassword(password.toCharArray());

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/channel/mqtt/Test.java

@@ -34,7 +34,7 @@
    public static void main(String[] args) {
        try{
            // åˆå§‹åŒ–连接池
            pool = new MqttClientPool("tcp://" + mqSvIp + ":" + mqSvPort, mqSvUserName, mqSvUserPassword, maxConnections);
            pool = new MqttClientPool("tcp://" + mqSvIp + ":" + mqSvPort, mqSvUserName, mqSvUserPassword, maxConnections, true);
            MqttClient clientSub = pool.popClient() ;
            testSubscribe(clientSub, topic1);
            testSubscribe(clientSub, topic2);

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol/p206V202404/parse/global/GlParse.java

@@ -328,14 +328,16 @@
        index += 8 ;
        cdData.orderNo = ByteUtil.BCD2String_BE(bs, index, index + 7) ;

        index += 8 ;
        cdData.startDt  = GlParse.parseRtuDt(bs, index) ;

        if(cdData.clResult == (byte)0x81){
            //2025-06-11 çŽ‹æ±Ÿæµ·å®‰æŽ’
            //失败了，下面就没有数据了
            return ;
        }

        index += 8 ;
        cdData.startDt  = GlParse.parseRtuDt(bs, index) ;


        index += 6 ;
        cdData.endDt  = GlParse.parseRtuDt(bs, index) ;


 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/MqttMsgParser.java

@@ -21,7 +21,7 @@
                vo.orgTag = topicGrp[0] ;
                vo.protocol = topicGrp[1] ;
                vo.devId = topicGrp[2] ;
                vo.topic = topicGrp[3] ;
                vo.name = topicGrp[3] ;
                return vo ;
            }
        }else{
@@ -30,7 +30,7 @@
    }

    public static String createPubTopic(MqttTopic tp) throws Exception {
        return tp.orgTag + "/" + tp.protocol + "/" + tp.devId + "/" + tp.topic ;
        return tp.orgTag + "/" + tp.protocol + "/" + tp.devId + "/" + tp.name;
    }

    public static MqttSubMsg parseSubMsg(MqttTopic subTopic, MqttMessage mqttMsg, MqttCallback callback) throws Exception {

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/MqttPubMsg.java

@@ -12,7 +12,7 @@

    public String mqttResultSendWebUrl ;//Mqtt返回命令结果 å‘向目的地web URL

    public String topic ;//消息主题
    public MqttTopic topic ;//消息主题
    public String msg ;//消息

    public boolean isCacheForOffLine ;//下行命令控制，消息中间件不在线是否缓存命令

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/MqttSubMsg.java

@@ -15,8 +15,8 @@
    public String deviceId ;//设备ID
    public String protocol;//协议

    public String topic ;//消息主题
    public String msg ;//消息
    public MqttTopic topic ;//消息主题
    public String metaData;//MQTT推送来的元数据

    public abstract boolean valid();


 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/MqttTopic.java

@@ -16,6 +16,18 @@
    public String orgTag ;//组织标识
    public String protocol ;//协议名称
    public String devId ;//设备（FBox）ID
    public String topic ;//消息主题
    public String name;//消息主题末端名称

    public boolean isEmpty(){
        return orgTag == null || protocol == null || devId == null || name == null
                || orgTag.trim().length() == 0 || protocol.trim().length() == 0 || devId.trim().length() == 0 || name.trim().length() == 0 ;
    }

    public String shortName(){
        return name;
    }

    public String longName(){
        return orgTag + "/" + protocol + "/" + devId + "/" + name;
    }
}

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/CodeSdV1.java

@@ -10,4 +10,5 @@
    public static final String cd_Stir = "01" ;//搅拌启停命令
    public static final String cd_Inject = "02" ;//注肥启停命令
    public static final String cd_Irr = "03" ;//灌溉启停命令
    public static final String cd_Param = "10" ;//设定参数
}

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/MqttSubMsgSdV1.java

@@ -24,8 +24,8 @@
    public MqttSubMsgSdV1(MqttTopic subTopic, String msg) {
        this.deviceId = subTopic.devId ;
        this.protocol = subTopic.protocol ;
        this.topic = subTopic.topic ;
        this.msg = msg ;
        this.topic = subTopic ;
        this.metaData = msg ;
    }
    public String toString(){
        StringBuilder sb = new StringBuilder();
@@ -35,24 +35,25 @@
                    .append("\n") ;
        }
        sb.append("主题:")
                .append(topic)
                .append("\n") ;
        sb.append("消息:")
                .append(msg)
                .append(topic.longName())
                .append("\n") ;
        if(vo4Up != null){
            sb.append("数据:")
                    .append(vo4Up.toString())
                    .append("\n") ;
        }else{
            sb.append("元数据:")
                    .append(metaData)
                    .append("\n") ;
        }

        return sb.toString() ;
    }

    public boolean subMsgMatchPubMsg(MqttPubMsg pubMsg){
        if (pubMsg instanceof MqttPubMsgSdV1) {
            MqttPubMsgSdV1 pubMsgSdV1 = (MqttPubMsgSdV1) pubMsg;
            //MqttPubMsgSdV1 pubMsgSdV1 = (MqttPubMsgSdV1) pubMsg;
            if(this.vo4Up != null && this.vo4Up instanceof StateVo){
                //只要上报的是状态数据，说明设备响应了命令
                return true ;
            }
        }
@@ -64,7 +65,7 @@
        if (topic == null || topic.isEmpty()) {
            return false;
        }
        if (msg == null || msg.isEmpty()) {
        if (metaData == null || metaData.isEmpty()) {
            return false;
        }
        return true;

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/ProtocolConstantSdV1.java

@@ -20,5 +20,6 @@
    public static final String PubTopicStir = "ctrlStir" ;//搅拌启停
    public static final String PubTopicInject = "ctrlInject" ;//注肥启停
    public static final String PubTopicIrr = "ctrlIrr" ;//灌溉启停
    public static final String PubTopicParam = "setParam" ;//设置参数

}

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/ProtocolParserSdV1.java

@@ -4,14 +4,16 @@
import com.alibaba.fastjson2.JSONObject;
import com.dy.common.mw.protocol.Command;
import com.dy.common.mw.protocol4Mqtt.MqttCallback;
import com.dy.common.mw.protocol4Mqtt.MqttMsgParser;
import com.dy.common.mw.protocol4Mqtt.MqttTopic;
import com.dy.common.mw.protocol4Mqtt.Vo4Up;
import com.dy.common.mw.protocol4Mqtt.pSdV1.comParam.ComCtrlVo;
import com.dy.common.mw.protocol4Mqtt.pSdV1.comParam.ComSetParamVo;
import com.dy.common.mw.protocol4Mqtt.pSdV1.downVos.FaultClearVo;
import com.dy.common.mw.protocol4Mqtt.pSdV1.downVos.InjectStartVo;
import com.dy.common.mw.protocol4Mqtt.pSdV1.downVos.ParamSetVo;
import com.dy.common.mw.protocol4Mqtt.pSdV1.downVos.StirStartVo;
import com.dy.common.mw.protocol4Mqtt.pSdV1.upVos.*;
import com.dy.common.mw.protocol4Mqtt.status.DevRunInfo;
import org.eclipse.paho.client.mqttv3.MqttMessage;

/**
@@ -21,38 +23,46 @@
 */
public class ProtocolParserSdV1 {
    public MqttSubMsgSdV1 parseSubMsg(MqttTopic subTopic, MqttMessage mqttMsg, MqttCallback callback) throws Exception {
        String msg = new String(mqttMsg.getPayload(), "UTF-8");
        if(JSON.isValid(msg)){
            throw new Exception("接收到MQTT消息，协议" + subTopic.protocol + "，设备ID" + subTopic.devId + "，主题" + subTopic.topic + "消息格式非json数据(" + msg + ")") ;
        String strTxt = new String(mqttMsg.getPayload(), "UTF-8");
        if(!JSON.isValid(strTxt)){
            throw new Exception("接收到MQTT消息，协议" + subTopic.protocol + "，设备ID" + subTopic.devId + "，主题" + subTopic.longName() + "消息格式非json数据(" + strTxt + ")") ;
        }
        MqttSubMsgSdV1 ms = new MqttSubMsgSdV1(subTopic, msg);
        MqttSubMsgSdV1 msg = new MqttSubMsgSdV1(subTopic, strTxt);
        Vo4Up vo ;
        switch (subTopic.topic) {
        DevRunInfo stInfo = null ;
        switch (subTopic.name) {
            case ProtocolConstantSdV1.SubTopicWeather -> {
                vo = JSON.parseObject(msg, WeatherVo.class);
                vo = JSON.parseObject(strTxt, WeatherVo.class);
                break;
            }
            case ProtocolConstantSdV1.SubTopicSoil -> {
                vo = JSON.parseObject(msg, SoilVo.class);
                vo = JSON.parseObject(strTxt, SoilVo.class);
                break;
            }
            case ProtocolConstantSdV1.SubTopicManure -> {
                vo = JSON.parseObject(msg, ManureVo.class);
                vo = JSON.parseObject(strTxt, ManureVo.class);
                break;
            }
            case ProtocolConstantSdV1.SubTopicState -> {
                //此处未完成，应该产生一些通信的info，供下面callback.notify(objs)通知出去
                vo = JSON.parseObject(msg, StateVo.class);
                vo = JSON.parseObject(strTxt, StateVo.class);
                StateVo stVo = (StateVo)vo ;
                stInfo = new DevRunInfo() ;
                stInfo.devId = msg.deviceId ;
                stInfo.stirRunning = (stVo.stirRunning==null?false:(stVo.stirRunning.byteValue()==1?true:false)) ; //搅拌运行 true是 false否
                stInfo.injectRunning = (stVo.injectRunning==null?false:(stVo.injectRunning.byteValue()==1?true:false)) ; //注肥运行 true是 false否
                stInfo.irrRunning = (stVo.irrRunning==null?false:(stVo.irrRunning.byteValue()==1?true:false)) ; //灌溉运行 true是 false否
                stInfo.alarm = (stVo.alarm==null?false:(stVo.alarm.byteValue()==1?true:false)) ; //报警 true是 false否
                break;
            }
            default -> {
                throw new Exception("接收到MQTT消息，协议" + subTopic.protocol + "，设备ID" + subTopic.devId + "，主题" + subTopic.topic + "消息解析逻辑未实现");
                throw new Exception("接收到MQTT消息，协议" + subTopic.protocol + "，设备ID" + subTopic.devId + "，主题" + subTopic.name + "消息解析逻辑未实现");
            }
        }
        ms.vo4Up = vo ;
        callback.callback(ms);
        callback.notify(null);//此处未完成
        return ms;
        msg.vo4Up = vo ;
        callback.callback(msg);
        callback.notify(msg.deviceId, stInfo);
        return msg;
    }

    public MqttPubMsgSdV1 createPubMsg(String orgTag, Command com) throws Exception {
@@ -86,6 +96,13 @@
                msg = this.createPubMsgOfIrr(orgTag, com);
                break;
            }
            case CodeSdV1.cd_Param -> {
                //设置参数
                this.checkParam(com);
                this.checkRtnWebUrl(com);
                msg = this.createPubMsgOfParam(orgTag, com);
                break;
            }
            default -> {
                throw new Exception("接收到MQTT命令，协议" + com.protocol + "版本" + com.protocolVersion + "功能码" + com.code + "构造器未实现");
            }
@@ -114,8 +131,8 @@
        msg.isCacheForOffLine = false ;
        msg.hasResponse = true ;
        msg.cd = CodeSdV1.cd_Fault ;
        msg.topic = MqttMsgParser.createPubTopic(new MqttTopic(orgTag, com.protocol, com.rtuAddr, ProtocolConstantSdV1.PubTopicFault)) ;
        msg.msg = JSON.toJSONString(new FaultClearVo(cvo.isDo)) ;
        msg.topic = new MqttTopic(orgTag, com.protocol, com.rtuAddr, ProtocolConstantSdV1.PubTopicFault) ;
        msg.msg = JSON.toJSONString(new FaultClearVo(cvo.startTrueStopFalse ?(byte)1:0)) ;
        return msg ;
    }
    private MqttPubMsgSdV1 createPubMsgOfStir(String orgTag, Command com) throws Exception {
@@ -130,8 +147,8 @@
        msg.isCacheForOffLine = false ;
        msg.hasResponse = true ;
        msg.cd = CodeSdV1.cd_Fault ;
        msg.topic = MqttMsgParser.createPubTopic(new MqttTopic(orgTag, com.protocol, com.rtuAddr, ProtocolConstantSdV1.PubTopicFault)) ;
        msg.msg = JSON.toJSONString(new StirStartVo(cvo.isDo)) ;
        msg.topic = new MqttTopic(orgTag, com.protocol, com.rtuAddr, ProtocolConstantSdV1.PubTopicStir) ;
        msg.msg = JSON.toJSONString(new StirStartVo(cvo.startTrueStopFalse ?(byte)1:0)) ;
        return msg ;
    }
    private MqttPubMsgSdV1 createPubMsgOfInject(String orgTag, Command com) throws Exception {
@@ -146,8 +163,8 @@
        msg.isCacheForOffLine = false ;
        msg.hasResponse = true ;
        msg.cd = CodeSdV1.cd_Fault ;
        msg.topic = MqttMsgParser.createPubTopic(new MqttTopic(orgTag, com.protocol, com.rtuAddr, ProtocolConstantSdV1.PubTopicFault)) ;
        msg.msg = JSON.toJSONString(new InjectStartVo(cvo.isDo)) ;
        msg.topic = new MqttTopic(orgTag, com.protocol, com.rtuAddr, ProtocolConstantSdV1.PubTopicInject) ;
        msg.msg = JSON.toJSONString(new InjectStartVo(cvo.startTrueStopFalse ?(byte)1:0)) ;
        return msg ;
    }
    private MqttPubMsgSdV1 createPubMsgOfIrr(String orgTag, Command com) throws Exception {
@@ -162,8 +179,25 @@
        msg.isCacheForOffLine = false ;
        msg.hasResponse = true ;
        msg.cd = CodeSdV1.cd_Fault ;
        msg.topic = MqttMsgParser.createPubTopic(new MqttTopic(orgTag, com.protocol, com.rtuAddr, ProtocolConstantSdV1.PubTopicFault)) ;
        msg.msg = JSON.toJSONString(new StirStartVo(cvo.isDo)) ;
        msg.topic = new MqttTopic(orgTag, com.protocol, com.rtuAddr, ProtocolConstantSdV1.PubTopicIrr) ;
        msg.msg = JSON.toJSONString(new StirStartVo(cvo.startTrueStopFalse ?(byte)1:0)) ;
        return msg ;
    }

    private MqttPubMsgSdV1 createPubMsgOfParam(String orgTag, Command com) throws Exception {
        JSONObject obj = (JSONObject) com.param;
        String json = obj.toJSONString();
        ComSetParamVo cvo = JSON.parseObject(json, ComSetParamVo.class);
        if(cvo == null){
            throw new Exception("json转ComSetParamVo为null") ;
        }
        MqttPubMsgSdV1 msg = new MqttPubMsgSdV1() ;
        this.setPubMsgBase(com, msg);
        msg.isCacheForOffLine = false ;
        msg.hasResponse = false ;
        msg.cd = CodeSdV1.cd_Param ;
        msg.topic = new MqttTopic(orgTag, com.protocol, com.rtuAddr, ProtocolConstantSdV1.PubTopicParam) ;
        msg.msg = JSON.toJSONString(new ParamSetVo(cvo.stirDuration, cvo.injectDuration)) ;
        return msg ;
    }


 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/comParam/ComCtrlVo.java

@@ -6,7 +6,7 @@
 * @Description
 */
public class ComCtrlVo {
    //是否控制动作，true是，false否
    //启停动作，true是，false否
    //可以执行功能码 00，01，02，03的动作
    public boolean isDo;//
    public boolean startTrueStopFalse;//
}

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/comParam/ComSetParamVo.java

New file
@@ -0,0 +1,23 @@
package com.dy.common.mw.protocol4Mqtt.pSdV1.comParam;

import com.alibaba.fastjson2.annotation.JSONField;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * @Author: liurunyu
 * @Date: 2025/6/11 17:03
 * @Description
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
public class ComSetParamVo {
    // æ…拌设定时间
    public Integer stirDuration ;

    // æ³¨è‚¥è®¾å®šæ—¶é—´
    public Integer injectDuration ;

}

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/downVos/FaultClearVo.java

@@ -16,10 +16,10 @@
@AllArgsConstructor
public class FaultClearVo implements Vo4Down {
    @JSONField(name = "故障解除")
    public boolean isDo ;
    public Byte isDo ;//1是，0否

    @Override
    public String toString(){
        return "故障解除：" + (isDo?"是":"否") ;
        return "故障解除：" + (isDo==1?"是":"否") ;
    }
}

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/downVos/InjectStartVo.java

@@ -16,9 +16,10 @@
@AllArgsConstructor
public class InjectStartVo implements Vo4Down {
    @JSONField(name = "注肥启停")
    public boolean isDo ;//true为启，false为停
    public Byte isDo ;//1是，0否

    @Override
    public String toString(){
        return "注肥启停：" + (isDo?"启":"停") ;
        return "注肥启停：" + (isDo==1?"启":"停") ;
    }
}

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/downVos/IrrStartVo.java

@@ -16,9 +16,10 @@
@AllArgsConstructor
public class IrrStartVo implements Vo4Down {
    @JSONField(name = "灌溉启停")
    public boolean isDo ;//true为启，false为停
    public Byte isDo ;//1是，0否

    @Override
    public String toString(){
        return "灌溉启停：" + (isDo?"启":"停") ;
        return "灌溉启停：" + (isDo==1?"是":"否") ;
    }
}

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/downVos/ParamSetVo.java

New file
@@ -0,0 +1,31 @@
package com.dy.common.mw.protocol4Mqtt.pSdV1.downVos;

import com.alibaba.fastjson2.annotation.JSONField;
import com.dy.common.mw.protocol4Mqtt.Vo4Down;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * @Author: liurunyu
 * @Date: 2025/6/11 16:55
 * @Description
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
public class ParamSetVo implements Vo4Down {
    @JSONField(name = "搅拌设定时间")
    public Integer stirDuration ;

    @JSONField(name = "注肥设定时间")
    public Integer injectDuration ;

    @Override
    public String toString(){
        StringBuilder sb = new StringBuilder();
        sb.append("搅拌设定时间：" + stirDuration + "\n" );
        sb.append("注肥设定时间：" + injectDuration + "\n" );
        return sb.toString();
    }
}

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/downVos/StirStartVo.java

@@ -16,9 +16,10 @@
@AllArgsConstructor
public class StirStartVo implements Vo4Down {
    @JSONField(name = "搅拌启停")
    public boolean isDo ;//true为启，false为停
    public Byte isDo ;//1是，0否

    @Override
    public String toString(){
        return "搅拌启停：" + (isDo?"启":"停") ;
        return "搅拌启停：" + (isDo==1?"是":"否") ;
    }
}

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/upVos/ManureVo.java

@@ -12,6 +12,18 @@
 */
@Data
public class ManureVo implements Vo4Up {
    @JSONField(name = "flexem_message_id")
    public Integer messageId ;//消息ID

    @JSONField(name = "肥料流量")
    public Float manureFlow ;//肥料流量

    @JSONField(name = "注肥时长")
    public Integer manureTime ;//注肥时长

    @JSONField(name = "搅拌时长")
    public Integer stirTime ;//搅拌时长

    @JSONField(name = "flexem_timestamp")
    public Long devDt ;//设备时间

@@ -27,9 +39,13 @@
    @Override
    public String toString(){
        StringBuilder sb = new StringBuilder();
        sb.append("水肥数据：") ;
        sb.append(" è®¾å¤‡æ—¶é—´ï¼š"+devDt) ;
        sb.append(" è®¾å¤‡æ—¶é—´ï¼š"+ this.getDevDtStr()) ;
        sb.append("水肥数据=>") ;
        sb.append(" æ¶ˆæ¯ID：" + messageId + ", ") ;
        sb.append(" è‚¥æ–™æµé‡ï¼š" + manureFlow + ", ") ;
        sb.append(" æ³¨è‚¥æ—¶é•¿ï¼š" + manureTime + ", ") ;
        sb.append(" æ…拌时长：" + stirTime + ", ") ;
        sb.append(" è®¾å¤‡æ—¶é—´ï¼š" + devDt + ", ") ;
        sb.append(" è®¾å¤‡æ—¶é—´ï¼š" +  this.getDevDtStr() + ", ") ;
        sb.append("\n") ;
        return sb.toString() ;
    }

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/upVos/SoilVo.java

@@ -12,6 +12,33 @@
 */
@Data
public class SoilVo implements Vo4Up {
    @JSONField(name = "flexem_message_id")
    public Integer messageId ;//消息ID

    @JSONField(name = "土壤湿度1")
    public Float soilHumidity1 ;//土壤湿度1

    @JSONField(name = "土壤湿度2")
    public Float soilHumidity2 ;//土壤湿度2

    @JSONField(name = "土壤湿度3")
    public Float soilHumidity3 ;//土壤湿度3

    @JSONField(name = "土壤湿度4")
    public Float soilHumidity4 ;//土壤湿度4

    @JSONField(name = "土壤湿度1")
    public Float soilTemperature1 ;//土壤温度1

    @JSONField(name = "土壤温度2")
    public Float soilTemperature2 ;//土壤温度2

    @JSONField(name = "土壤温度3")
    public Float soilTemperature3 ;//土壤温度3

    @JSONField(name = "土壤温度4")
    public Float soilTemperature4 ;//土壤温度4

    @JSONField(name = "flexem_timestamp")
    public Long devDt ;//设备时间

@@ -27,10 +54,20 @@
    @Override
    public String toString(){
        StringBuilder sb = new StringBuilder();
        sb.append("墒情数据：") ;
        sb.append(" è®¾å¤‡æ—¶é—´ï¼š"+devDt) ;
        sb.append(" è®¾å¤‡æ—¶é—´ï¼š"+ this.getDevDtStr()) ;
        sb.append("墒情数据=>") ;
        sb.append(" æ¶ˆæ¯ID：" + messageId + ", ") ;
        sb.append(" åœŸå£¤æ¹¿åº¦1：" + soilHumidity1 + ", ") ;
        sb.append(" åœŸå£¤æ¹¿åº¦2：" + soilHumidity2 + ", ") ;
        sb.append(" åœŸå£¤æ¹¿åº¦3：" + soilHumidity3 + ", ") ;
        sb.append(" åœŸå£¤æ¹¿åº¦4：" + soilHumidity4 + ", ") ;
        sb.append(" åœŸå£¤æ¸©åº¦1：" + soilTemperature1 + ", ") ;
        sb.append(" åœŸå£¤æ¸©åº¦2：" + soilTemperature2 + ", ") ;
        sb.append(" åœŸå£¤æ¸©åº¦3：" + soilTemperature3 + ", ") ;
        sb.append(" åœŸå£¤æ¸©åº¦4：" + soilTemperature4 + ", ") ;
        sb.append(" è®¾å¤‡æ—¶é—´ï¼š" + devDt + ", ") ;
        sb.append(" è®¾å¤‡æ—¶é—´ï¼š" +  this.getDevDtStr() + ", ") ;
        sb.append("\n") ;
        return sb.toString() ;

    }
}

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/upVos/StateVo.java

@@ -12,6 +12,21 @@
 */
@Data
public class StateVo implements Vo4Up {
    @JSONField(name = "flexem_message_id")
    public Integer messageId ;//消息ID

    @JSONField(name = "搅拌运行")
    public Byte stirRunning ;//搅拌运行

    @JSONField(name = "注肥运行")
    public Byte injectRunning ;//注肥运行

    @JSONField(name = "灌溉运行")
    public Byte irrRunning ;//灌溉运行

    @JSONField(name = "报警")
    public Byte alarm ;//报警

    @JSONField(name = "flexem_timestamp")
    public Long devDt ;//设备时间

@@ -27,9 +42,14 @@
    @Override
    public String toString(){
        StringBuilder sb = new StringBuilder();
        sb.append("状态数据：") ;
        sb.append(" è®¾å¤‡æ—¶é—´ï¼š"+devDt) ;
        sb.append(" è®¾å¤‡æ—¶é—´ï¼š"+ this.getDevDtStr()) ;
        sb.append("状态数据=>") ;
        sb.append(" æ¶ˆæ¯ID：" + messageId + ", ") ;
        sb.append(" æ…拌运行：" + stirRunning + ", ") ;
        sb.append(" æ³¨è‚¥è¿è¡Œï¼š" + injectRunning + ", ") ;
        sb.append(" çŒæº‰è¿è¡Œï¼š" + irrRunning + ", ") ;
        sb.append(" æŠ¥è­¦ï¼š" + alarm + ", ") ;
        sb.append(" è®¾å¤‡æ—¶é—´ï¼š" + devDt + ", ") ;
        sb.append(" è®¾å¤‡æ—¶é—´ï¼š" +  this.getDevDtStr() + ", ") ;
        sb.append("\n") ;
        return sb.toString() ;
    }

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/pSdV1/upVos/WeatherVo.java

@@ -43,7 +43,7 @@

    public String devDtStr ;//设备时间
    public String getDevDtStr() {
        if(devDt == null){
        if(devDt != null){
            return DateTime.yyyy_MM_dd_HH_mm_ss(DateTime.getDate(devDt)) ;
        }else{
            return "" ;
@@ -53,17 +53,17 @@
    @Override
    public String toString(){
        StringBuilder sb = new StringBuilder();
        sb.append("气象数据：") ;
        sb.append(" æ¶ˆæ¯ID："+messageId) ;
        sb.append(" äºŒæ°§åŒ–碳："+carbonDioxide) ;
        sb.append(" å…‰ç…§å¼ºåº¦ï¼š"+lightIntensity) ;
        sb.append(" å¤§æ°”压力："+atmosphericPressure) ;
        sb.append(" ç©ºæ°”温度："+airTemperature) ;
        sb.append(" ç©ºæ°”湿度："+airHumidity) ;
        sb.append(" PM2.5："+pm25) ;
        sb.append(" PM10："+pm10) ;
        sb.append(" è®¾å¤‡æ—¶é—´ï¼š"+devDt) ;
        sb.append(" è®¾å¤‡æ—¶é—´ï¼š"+ this.getDevDtStr()) ;
        sb.append("气象数据=>") ;
        sb.append(" æ¶ˆæ¯ID：" + messageId + ", ") ;
        sb.append(" äºŒæ°§åŒ–碳：" + carbonDioxide + ", ") ;
        sb.append(" å…‰ç…§å¼ºåº¦ï¼š" + lightIntensity + ", ") ;
        sb.append(" å¤§æ°”压力：" + atmosphericPressure + ", ") ;
        sb.append(" ç©ºæ°”温度：" + airTemperature + ", ") ;
        sb.append(" ç©ºæ°”湿度：" + airHumidity + ", ") ;
        sb.append(" PM2.5：" + pm25 + ", ") ;
        sb.append(" PM10：" + pm10 + ", ") ;
        sb.append(" è®¾å¤‡æ—¶é—´ï¼š" + devDt + ", ") ;
        sb.append(" è®¾å¤‡æ—¶é—´ï¼š" +  this.getDevDtStr() + ", ") ;
        sb.append("\n") ;
        return sb.toString() ;
    }

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/status/DevOnLineInfo.java

File was renamed from pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/status/DevOnLineSt.java
@@ -9,7 +9,7 @@
 * @Description
 */
@Data
public class DevOnLineSt implements MqttNotifyInfo {
public class DevOnLineInfo implements MqttNotifyInfo {
    public String id ;
    public String protocol ;
    public Boolean onLine ;

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/status/DevRunInfo.java

File was renamed from pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/protocol4Mqtt/status/DevRunSt.java
@@ -9,8 +9,8 @@
 * @Description
 */
@Data
public class DevRunSt implements MqttNotifyInfo {
    public String id ;
public class DevRunInfo implements MqttNotifyInfo {
    public String devId ;//MQTT设置ID
    public Boolean stirRunning ;//搅拌运行 true是 false否
    public Boolean injectRunning ;//注肥运行 true是 false否
    public Boolean irrRunning ;//灌溉运行 true是 false否

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/Server.java

@@ -474,6 +474,7 @@
                if(mqVo.poolMaxSize <= 1 || mqVo.poolMaxSize > 1000){
                    throw new Exception("config.mqtt.poolMaxSize配置的连接池连接最大数量不合法") ;
                }
                mqVo.useMemoryPersistence = conf.getSetAttrBoolean(doc, "config.mqtt", "useMemoryPersistence", null, null) ;
                String proAndDevIds = conf.getSetAttrTxt(doc, "config.mqtt", "protocolAndDeviceIds", null, false, null) ;
                if(proAndDevIds == null || proAndDevIds.trim().equals("")){
                    throw new Exception("config.mqtt.protocolAndDeviceIds配置不合法") ;

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/ServerShutDownHook.java

@@ -32,7 +32,7 @@
                com2.code = CodeLocal.stopMqttSv ;
                com2.type = CommandType.innerCommand ;
                new CommandInnerDeaLer().deal(com2) ;

                log.info("关闭程序前，关闭了MQTT服务");
            }catch (Exception e){
                log.error("程序（控制台）关闭钩子发生异常", e);
            }

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/server/local/CommandInnerDeaLer.java

@@ -113,6 +113,10 @@
                rCom = this.stopMqttSv(com);
                break;
            }
            case CodeLocal.recoverMqttSv -> {
                rCom = this.recoverMqttSv(com);
                break;
            }
            default -> {
                rCom = ReturnCommand.errored("出错，收到内部命令的功能码不能识别！", com.getId(), com.getCode());
                break;

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/server/local/ReturnCommand.java

@@ -19,7 +19,6 @@
     * @param message
     */
    public static Command successed(String message, String commandId, String code, Object attachment) {
        log.info(message);
        Command command = new Command().createReturnSuccessCommand(message, commandId, code);
        command.setAttachment(attachment);
        return command;
@@ -29,7 +28,6 @@
     * @param message
     */
    public static Command successed(String message, String commandId, String code) {
        log.info(message);
        return new Command().createReturnSuccessCommand(message, commandId, code);
    }
    /**
@@ -37,7 +35,6 @@
     * @param message
     */
    public static Command errored(String message, String commandId, String code) {
        log.error(message);
        return new Command().createReturnErrorCommand(message, commandId, code);
    }
}

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/server/local/localProtocol/CodeLocal.java

@@ -38,7 +38,7 @@

    public static final String onPartLineMqtt = "LMCD0002" ;//查询部分MQTT设备在线情况

    public static final String onLineStatisticsMqtt = "LMCD0003" ;//查询所有MQTT设备状态统计情况
    public static final String onLineStatisticsMqtt = "LMCD0003" ;//查询所有MQTT设备在线状态统计情况

    public static final String allRtuStatesMqtt = "LMCD0010" ;//查询所有MQTT设备状态

@@ -48,5 +48,7 @@

    public static final String stopMqttSv = "LMCD0110" ;//停止Mqtt服务

    public static final String recoverMqttSv = "LMCD0112" ;//重启MQTT服务，接入新的MQTT连接


}

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

@@ -1,6 +1,6 @@
package com.dy.rtuMw.server.mqtt;

import com.dy.common.mw.protocol4Mqtt.status.DevRunSt;
import com.dy.common.mw.protocol4Mqtt.status.DevRunInfo;
import com.dy.rtuMw.server.forTcp.RtuLogDealer;
import com.dy.rtuMw.server.local.localProtocol.RtuOnLineStateStatisticsVo;

@@ -55,7 +55,7 @@
            Map.Entry<String, DevStatus> entry = null ;
            while(it.hasNext()){
                entry = it.next() ;
                if(((DevStatus)entry).onLine != null && ((DevStatus)entry).onLine.booleanValue()){
                if((entry.getValue()).onLine != null && (entry.getValue()).onLine.booleanValue()){
                    vo.onLineNum++ ;
                }else{
                    vo.offLineNum++ ;
@@ -166,7 +166,7 @@
        }
    }

    public static void setStatus(String devId, DevRunSt st){
    public static void setStatus(String devId, DevRunInfo st){
        DevStatus vo = map.get(devId) ;
        if(vo != null) {
            if(st.stirRunning != null){

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

@@ -3,12 +3,15 @@
import com.dy.common.mw.channel.mqtt.MqttClientPool;
import com.dy.common.mw.protocol4Mqtt.MqttNotify;
import com.dy.common.mw.protocol4Mqtt.MqttNotifyInfo;
import com.dy.common.mw.protocol4Mqtt.status.DevOnLineSt;
import com.dy.common.mw.protocol4Mqtt.status.DevRunSt;
import com.dy.common.mw.protocol4Mqtt.status.DevOnLineInfo;
import com.dy.common.mw.protocol4Mqtt.status.DevRunInfo;
import com.dy.rtuMw.server.ServerProperties;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.eclipse.paho.client.mqttv3.MqttClient;

import java.util.ArrayList;
import java.util.List;

/**
 * @Author: liurunyu
@@ -24,6 +27,8 @@
    private MqttUnitConfigVo configVo ;

    private MqttClientPool pool;

    private List<MqttClient> subClients ;

    private MqttManager(){
    }
@@ -43,8 +48,9 @@
     * @throws Exception
     */
    public void start()throws Exception{
        subClients = new ArrayList<>();
        String URL = "tcp://" + this.configVo.svIp + ":" + this.configVo.svPort;
        this.pool = new MqttClientPool(URL, this.configVo.svUserName, this.configVo.svUserPassword, this.configVo.poolMaxSize);
        this.pool = new MqttClientPool(URL, this.configVo.svUserName, this.configVo.svUserPassword, this.configVo.poolMaxSize, this.configVo.useMemoryPersistence);
        if(this.pool.isClose()){
            throw new Exception("Mqtt连接池初始化失败");
        }
@@ -57,6 +63,7 @@
        if(clientSub == null || !clientSub.isConnected()){
            throw new Exception("Mqtt连接池获得订阅连接不可用");
        }
        subClients.add(clientSub) ;
        // è®¢é˜…主题
        for(int i = 0; i < this.configVo.subTopics.length; i++){
            for(int j = 0 ; j < this.configVo.protocolAndDeviceIds.length; j++){
@@ -70,15 +77,15 @@
                            public void notify(String devId, MqttNotifyInfo... infos) {
                                if(devId != null && infos != null && infos.length > 0){
                                    for(MqttNotifyInfo info : infos){
                                        if(info instanceof DevOnLineSt){
                                            DevOnLineSt onLineSt = (DevOnLineSt)info;
                                        if(info instanceof DevOnLineInfo){
                                            DevOnLineInfo onLineSt = (DevOnLineInfo)info;
                                            if(onLineSt.onLine != null && onLineSt.onLine.booleanValue()){
                                                DevStatusDealer.onLine(devId, ((DevOnLineSt)info).protocol);
                                                DevStatusDealer.onLine(devId, ((DevOnLineInfo)info).protocol);
                                            }else{
                                                DevStatusDealer.offLine(devId);
                                            }
                                        } else if(info instanceof DevRunSt){
                                            DevStatusDealer.setStatus(devId, (DevRunSt)info);
                                        } else if(info instanceof DevRunInfo){
                                            DevStatusDealer.setStatus(devId, (DevRunInfo)info);
                                        }
                                    }
                                }
@@ -90,6 +97,18 @@
    }

    public void stop()throws Exception{
        if(subClients != null && subClients.size() > 0){
            for (MqttClient client : subClients) {
                if(client != null && client.isConnected()){
                    try{
                        client.disconnect();
                        client.close();
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }
            }
        }
        if(this.pool != null){
            // å…³é—­è¿žæŽ¥æ± 
            this.pool.close();

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

@@ -22,51 +22,29 @@

    @Override
    public void messageArrived(String topic, MqttMessage msg) throws Exception {
        MqttTopic subTopic = MqttMsgParser.parseSubTopic(topic) ;
        MqttSubMsg subMsg = MqttMsgParser.parseSubMsg(subTopic, msg, new MqttCallback(){
            @Override
            public void callback(MqttSubMsg subMsg) {
                DevStatusDealer.onLine(subMsg.deviceId, subMsg.protocol);
                DevStatusDealer.afterReceiveSubMessage(subMsg.deviceId);
                RtuLogDealer.log4Mqtt(subMsg.deviceId, "订阅消息    ä¸»é¢˜ï¼š" + subMsg.topic + "   æ¶ˆæ¯ï¼š" + subMsg.msg);
            }
            @Override
            public void notify(String devId, MqttNotifyInfo... infos) {
                if(notify != null){
                    notify.notify(devId, infos) ;
        try {
            MqttTopic subTopic = MqttMsgParser.parseSubTopic(topic);
            MqttSubMsg subMsg = MqttMsgParser.parseSubMsg(subTopic, msg, new MqttCallback() {
                @Override
                public void callback(MqttSubMsg subMsg) {
                    DevStatusDealer.onLine(subMsg.deviceId, subMsg.protocol);
                    DevStatusDealer.afterReceiveSubMessage(subMsg.deviceId);
                    RtuLogDealer.log4Mqtt(subMsg.deviceId, "订阅消息    ä¸»é¢˜ï¼š" + subMsg.topic.longName() + "   å…ƒæ•°æ®ï¼š" + subMsg.metaData);
                }
            }
        }) ;
        this.nextDeal(subMsg);
    }
    private void nextDeal(MqttSubMsg subMsg)throws Exception {
        subMsg.action(new Callback() {
            @Override
            public void call(Object obj) {
                MqttSubMsg subMs = (MqttSubMsg) obj ;
                MqttPubMsg pubMs = MqttPubMsgCache.matchFromTail(subMs) ;
                if(pubMs != null){
                    //匹配到下行消息（命令）
                    subMs.mqttResultSendWebUrl = pubMs.mqttResultSendWebUrl ;
                    subMs.commandId = pubMs.commandId ;
                    try {
                        MqttComResultCache.getInstance().cacheMqttComResult(new MqttComResultNode(subMs));
                    } catch (Exception e) {
                        log.error("缓存发布消息（命令）结果发生异常", e);

                @Override
                public void notify(String devId, MqttNotifyInfo... infos) {
                    if (notify != null) {
                        notify.notify(devId, infos);
                    }
                }
                try{
                    MqttSubMsgCache.getInstance().cacheMsg(new MqttSubMsgNode(subMsg));
                }catch (Exception e){
                    log.error("缓存订阅消息数据发生异常", e);
                }
            }
            @Override
            public void call(Object... objs) {
            }
            @Override
            public void exception(Exception e) {
            }
        });
            });
            this.nextDeal(subMsg);
        }catch(Exception e){
            log.error("处理MQTT订阅消息发生异常", e);
        }
    }
    private void nextDeal(MqttSubMsg subMsg)throws Exception {
        subMsg.action(new MqttSubMsgDealer());
    }
}

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

@@ -86,8 +86,7 @@
        while(node != null && node.obj != null){
            obj = (MqttPubMsgNode)node.obj;
            pubMsg = obj.result ;
            if(pubMsg != null
                    && subMsg.subMsgMatchPubMsg(pubMsg)){
            if(pubMsg != null && subMsg.subMsgMatchPubMsg(pubMsg)){
                obj.onceReceivedResult = true ;//标识已经收到命令结果
                return pubMsg;
            }else{

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

@@ -65,7 +65,7 @@
        }else{
            if(mqttClient != null && mqttClient.isConnected()){
                try {
                    mqttManager.publishMsg(mqttClient, this.result.topic, this.result.msg);
                    mqttManager.publishMsg(mqttClient, this.result.topic.longName(), this.result.msg);
                    DevStatusDealer.afterSendPubMessage(this.result.deviceId);
                    RtuLogDealer.log4Mqtt(this.result.deviceId, "发布消息    ä¸»é¢˜ï¼š" + this.result.topic + "   æ¶ˆæ¯ï¼š" + this.result.msg);
                    log.info("发布MQTT消息（主题=" + this.result.topic + "）" + this.result.msg);
@@ -74,7 +74,11 @@
                }finally {
                    mqttManager.pushMqttClient(mqttClient);
                }
                return false ;
                if(this.result.hasResponse){
                    return false ;
                }else{
                    return true ;
                }
            }else{
                //未曾连接MQTT服务器
                return this.decideRemoveNodeFromCach(now) ;

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

New file
@@ -0,0 +1,41 @@
package com.dy.rtuMw.server.mqtt;

import com.dy.common.mw.protocol4Mqtt.MqttPubMsg;
import com.dy.common.mw.protocol4Mqtt.MqttSubMsg;
import com.dy.common.util.Callback;
import lombok.extern.slf4j.Slf4j;

/**
 * @Author: liurunyu
 * @Date: 2025/6/11 17:33
 * @Description
 */
@Slf4j
public class MqttSubMsgDealer implements Callback {
    @Override
    public void call(Object obj) {
        MqttSubMsg subMs = (MqttSubMsg) obj ;
        MqttPubMsg pubMs = MqttPubMsgCache.matchFromTail(subMs) ;
        if(pubMs != null){
            //匹配到下行消息（命令）
            subMs.mqttResultSendWebUrl = pubMs.mqttResultSendWebUrl ;
            subMs.commandId = pubMs.commandId ;
            try {
                MqttComResultCache.getInstance().cacheMqttComResult(new MqttComResultNode(subMs));
            } catch (Exception e) {
                log.error("缓存发布消息（命令）结果发生异常", e);
            }
        }
        try{
            MqttSubMsgCache.getInstance().cacheMsg(new MqttSubMsgNode(subMs));
        }catch (Exception e){
            log.error("缓存订阅消息数据发生异常", e);
        }
    }
    @Override
    public void call(Object... objs) {
    }
    @Override
    public void exception(Exception e) {
    }
}

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

@@ -13,6 +13,7 @@
    public String svUserName ;//
    public String svUserPassword ;//
    public Integer poolMaxSize ;//
    public Boolean useMemoryPersistence ;
    public String[] protocolAndDeviceIds ;//设备协议与ID（FBox）id
    public String[] deviceIds ;//设备（FBox）id
    public String[] subTopics ;//订阅的主题
@@ -27,6 +28,7 @@
        this.svUserName = "dyyjy" ;
        this.svUserPassword = "Dyyjy2025,;.abc!@#" ;
        this.poolMaxSize = 10 ;
        useMemoryPersistence = true ;
        this.pubTopicQos = 1 ;
        this.noSubThenOff = 10 * 60 * 10000L ;
    }

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/server/rtuData/p206V1/TkDealAlarmStatus.java

@@ -63,13 +63,14 @@
                                    toNext = true ;
                                }
                            }
                            if(this.needSave2Db( d.getRtuAddr())){
                            //2025-06-11 å› ä¸º81命令是RTU状态改变时即上报，所以可以及时体现状态变化，进而不受存储控制
                            //if(this.needSave2Db( d.getRtuAddr())){
                                //在一个小时内没有存储报警数据
                                this.toMsCenter(controller, d.getRtuAddr(), dV1, dataCd81Vo.alarmVo, dataCd81Vo.stateVo);
                                this.saveOrUpdateLast(sv, controller, d.getRtuAddr(), dataCd81Vo.rtuDt, dV1, dataCd81Vo.alarmVo, dataCd81Vo.stateVo);
                                this.saveHistory(sv, controller, d.getRtuAddr(), dataCd81Vo.rtuDt, dV1, dataCd81Vo.alarmVo, dataCd81Vo.stateVo);
                                this.cacheSaveTime(d.getRtuAddr());
                            }
                                //this.cacheSaveTime(d.getRtuAddr());
                            //}
                        }
                    } else if (cdObj instanceof DataCd84Vo) {
                        //阀开工作报
@@ -78,7 +79,7 @@
                            Object[] objs = this.getTaskResults(TkPreGenObjs.taskId);
                            DbSv sv = (DbSv) objs[0];
                            PrController controller = (PrController) objs[1];
                            /** é˜€å¼€å·¥ä½œæŠ¥ï¼Œé˜€é—¨çŠ¶æ€ä¸€å®šæ˜¯æ‰“开状态，所以不设置下面数据以备后续任务设置虚拟卡状态
                            /* é˜€å¼€å·¥ä½œæŠ¥ï¼Œé˜€é—¨çŠ¶æ€ä¸€å®šæ˜¯æ‰“开状态，所以不设置下面数据以备后续任务设置虚拟卡状态
                             if(dataCd84Vo.stateVo != null && dataCd84Vo.stateVo.valveState != null && dataCd84Vo.stateVo.valveState.byteValue() == DataStateVo.ValveCloseState){
                                 if(controller != null && controller.getIntakeId() != null){
                                     this.taskResult = new Object[]{controller.getIntakeId(), DataStateVo.ValveCloseState} ;

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/server/rtuData/p206V2/TkDealAlarmStatusV2.java

@@ -76,7 +76,8 @@
                        //随机自报报
                        DataCd81Vo dataCd81Vo = (DataCd81Vo)cdObj ;
                        if (dataCd81Vo.alarmVo != null || dataCd81Vo.stateVo != null) {
                            if(this.needSave2Db( d.getRtuAddr())) {
                            //2025-06-11 å› ä¸º81命令是RTU状态改变时即上报，所以可以及时体现状态变化，进而不受存储控制
                            //if(this.needSave2Db( d.getRtuAddr())) {
                                //在一个小时内没有存储报警数据
                                Object[] objs = this.getTaskResults(TkPreGenObjsV2.taskId);
                                DbSv sv = (DbSv) objs[0];
@@ -91,7 +92,7 @@
                                        toNext = true ;
                                    }
                                }
                            }
                            //}
                        }
                    } else if (cdObj instanceof DataCd84Vo) {
                        //阀开工作报

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/server/rtuData/p206V202404/TkDealAlarmStatusV202404.java

@@ -48,18 +48,16 @@
            PrController controller = (PrController)objs[1] ;
            if(cdObj instanceof DataCd81Vo){
                //设备终端随机自报
                if(this.needSave2Db( d.getRtuAddr())) {
                    //在一个小时内没有存储报警数据
                    DataCd81Vo cdData = (DataCd81Vo)(cdObj) ;
                    this.toMsCenter(controller, d.getRtuAddr(), dV202404, cdData, dV202404.dt);
                    this.saveOrUpdateLast(sv, controller, d.getRtuAddr(), cdData.rtuDt, cdData, dV202404.dt) ;
                    this.saveHistory(sv, controller, d.getRtuAddr(), cdData.rtuDt, cdData, dV202404.dt) ;
                    this.cacheSaveTime(d.getRtuAddr());
                    if(cdData.statePump != null && cdData.statePump.byteValue() == AlarmVo.DevCloseState){
                        if(controller != null && controller.getIntakeId() != null){
                            this.taskResult = new Object[]{controller.getIntakeId(), AlarmVo.DevCloseState} ;
                            toNext = true ;
                        }
                //81功能码数据，不进行“在一个小时内没有存储报警数据”的控制
                DataCd81Vo cdData = (DataCd81Vo)(cdObj) ;
                this.toMsCenter(controller, d.getRtuAddr(), dV202404, cdData, dV202404.dt);
                this.saveOrUpdateLast(sv, controller, d.getRtuAddr(), cdData.rtuDt, cdData, dV202404.dt) ;
                this.saveHistory(sv, controller, d.getRtuAddr(), cdData.rtuDt, cdData, dV202404.dt) ;
                this.cacheSaveTime(d.getRtuAddr());
                if(cdData.statePump != null && cdData.statePump.byteValue() == AlarmVo.DevCloseState){
                    if(controller != null && controller.getIntakeId() != null){
                        this.taskResult = new Object[]{controller.getIntakeId(), AlarmVo.DevCloseState} ;
                        toNext = true ;
                    }
                }
            }else if(cdObj instanceof DataCd80_5BVo){
@@ -78,7 +76,9 @@
                        }
                    }
                }
            }else if(cdObj instanceof DataCd93_A3Vo){
            }
            /* 2025-06-11 çŽ‹æ±Ÿæµ·è¯´ï¼šä»¥ä¸‹æ•°æ®ä¸­çš„状态不准确，尤其阀门状态，所以不再采纳
            else if(cdObj instanceof DataCd93_A3Vo){
                //平台/APP远程关闭水泵/阀门   +  åˆ·å¡å…³æ³µ/阀上报
                if(this.needSave2Db( d.getRtuAddr())) {
                    //在一个小时内没有存储报警数据
@@ -127,6 +127,7 @@
                    }
                }
            }
             */
        }catch (Exception e){
            log.error("保存控制器报警和状态数据时发生异常", e);
        }

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/web/com/CommandCtrl.java

@@ -58,7 +58,21 @@
     */
    @GetMapping("/rtuLogFile")
    public void rtuLogFile(String rtuAddr, HttpServletRequest req, HttpServletResponse rep){
        File logFile = ResourceUnit.getInstance().getLogFile(rtuAddr + ".log") ;
        logFile(rtuAddr, req, rep) ;
    }

    /**
     * ä¸‹è½½æŽ§åˆ¶å™¨ï¼ˆRTU）上下行数据的log日志文件
     * @param devId
     * @param req
     * @param rep
     */
    @GetMapping("/mqttDevLogFile")
    public void mqttDevLogFile(String devId, HttpServletRequest req, HttpServletResponse rep){
        logFile(devId, req, rep) ;
    }
    private void logFile(String fileName, HttpServletRequest req, HttpServletResponse rep){
        File logFile = ResourceUnit.getInstance().getLogFile(fileName + ".log") ;
        if(logFile != null && logFile.exists()){
            //在Spring Boot中，application/octet-stream;charset=UTF-8通常表示响应的内容是字节流，
            //并且字符集是UTF-8。对于这种类型的响应，Spring Boot默认使用ByteArrayHttpMessageConverter来处理，
@@ -67,7 +81,7 @@
            //字符集通常用于文本内容，而application/octet-stream通常用于二进制内容，因此在这种情况下指定字符集可能是不合适的。
            //不过，如果你确实需要处理带有特定字符集的application/octet-stream响应，你可能需要自定义HttpMessageConverter。
            rep.addHeader("content-type", "application/octet-stream;charset=UTF-8");
            rep.addHeader("Content-Disposition", "attachment;fileName=" + (rtuAddr + ".log")) ;
            rep.addHeader("Content-Disposition", "attachment;fileName=" + (fileName + ".log")) ;
            ServletOutputStream out = null;
            FileInputStream in = null ;
            try {
@@ -112,27 +126,38 @@
     */
    @GetMapping("/rtuLogText")
    public BaseResponse<List<String>> rtuLogText(String rtuAddr){
        return logText(rtuAddr, true) ;
    }
    /**
     * ä¸‹è½½æŽ§åˆ¶å™¨ï¼ˆRTU）上下行数据的log日志文件
     * @param devId
     */
    @GetMapping("/mqttDevLogText")
    public BaseResponse<List<String>> mqttDevLogText(String devId){
        return logText(devId, false) ;
    }
    /**
     * ä¸‹è½½æŽ§åˆ¶å™¨ï¼ˆRTU）上下行数据的log日志文件
     * @param fileName
     * @param reverseOrder
     */
    private BaseResponse<List<String>> logText(String fileName, boolean reverseOrder){
        List<String> list ;
        File logFile = ResourceUnit.getInstance().getLogFile(rtuAddr + ".log") ;
        File logFile = ResourceUnit.getInstance().getLogFile(fileName + ".log") ;
        if(logFile != null && logFile.exists()){
            BufferedReader reader = null ;
            try {
                reader = new BufferedReader(new FileReader(logFile)) ;
                //新的实现方法
                Stream<String> linesStream = reader.lines() ;
                //list = linesStream.toList() ; //按原来顺序
                list = linesStream.sorted(Comparator.reverseOrder()).collect(Collectors.toList()) ;//倒序
                /* åŽŸæ¥çš„实现方法
                list = new ArrayList() ;
                String line ;
                while((line = reader.readLine()) != null){
                    list.add(line) ;
                if(reverseOrder){
                    list = linesStream.sorted(Comparator.reverseOrder()).collect(Collectors.toList()) ;//倒序
                }else{
                    list = linesStream.toList() ; //按原来顺序
                }
                */
                return BaseResponseUtils.buildSuccess(list);
            } catch (Exception e) {
                list = new ArrayList() ;
                list.add("读取控制器（" + rtuAddr + "）的日志文件异常：" + (e.getMessage() == null?"":("，" + e.getMessage()))) ;
                list.add("读取控制器（" + fileName + "）的日志文件异常：" + (e.getMessage() == null?"":("，" + e.getMessage()))) ;
                return BaseResponseUtils.buildSuccess(list);
            }finally{
                if(reader != null){
@@ -144,7 +169,7 @@
            }
        }else{
            list = new ArrayList() ;
            list.add("未得到控制器（" + rtuAddr + "）的日志文件") ;
            list.add("未得到控制器（" + fileName + "）的日志文件") ;
            return BaseResponseUtils.buildSuccess(list);
        }
    }

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/resources/config.properties

@@ -10,7 +10,7 @@
#   ç”˜å·žï¼š gz
#   å‡‰å·žï¼š lz
#   é‡‘川： jc
base.orgTag=mq
base.orgTag=ym

# 233服务器：
#   å…ƒè°‹ï¼š 60000
@@ -45,7 +45,7 @@
#   ç”˜å·žï¼š mqtt.enable=false  mqtt.protocolAndDeviceIds= mqtt.topicAndQos=
#   å‡‰å·žï¼š mqtt.enable=false  mqtt.protocolAndDeviceIds= mqtt.topicAndQos=
#   é‡‘川： mqtt.enable=true  mqtt.protocolAndDeviceIds=? mqtt.topicAndQos=weather,1;soil,1;manure,1;state,1
mqtt.enable=true
mqtt.enable=false
mqtt.protocolAndDeviceIds=sd1/338220031439,sd1/338220031440
mqtt.subTopicAndQos=weather,1;soil,1;manure,1;state,1
#MQtt设备在一定时间（分钟）后未发布消息，认为设备离线

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/resources/config.xml

@@ -171,6 +171,7 @@
    svUserName MQTT服务器用户名
    svUserPassword MQTT服务器用户密码
    poolMaxSize è¿žæŽ¥æ± æœ€å¤§è¿žæŽ¥æ•°
    useMemoryPersistence ä½¿ç”¨å†…存持久化而非默认的文件持久化(true是 false否)
    protocolAndDeviceIds åœ¨å­ç³»ç»Ÿï¼ˆorgTag）中接入的设备(FBox)所用协议及设备id集合,多个用逗号隔开，协议与ID用正斜杠隔开，例如：sd1/338220031439,sd1/338220031440
    subTopicAndQos: è®¢é˜…主题与Qos，主题名与其Qos用逗号隔开，多个主题及Qos用分号隔开，例如：ym/topic1,1;ym/topic2,1;ym/topic3,1，如果有多个OrgTag，主题前缀用其OrgTag
    pubTopicQos: å‘布主题的Qos，取值范围：
@@ -185,6 +186,7 @@
          svUserName="dyyjy"
          svUserPassword="Dyyjy2025,;.abc!@#"
          poolMaxSize="10"
          useMemoryPersistence="true"
          protocolAndDeviceIds="${mqtt.protocolAndDeviceIds}"
          subTopicAndQos="${mqtt.subTopicAndQos}"
          pubTopicQos="1"

 pipIrr-platform/pipIrr-mw/pipIrr-param-set/src/main/java/com/dy/pipIrrParamSet/paramSet/ParamKey.java

@@ -31,6 +31,8 @@
    public static final String TcpPort = "tcpPort" ;
    public static final String UpDataMinInterval = "upDataMinInterval" ;
    public static final String WebPort = "webPort" ;
    public static final String MqttEnable = "mqttEnable" ;
    public static final String MqttNoSubThenOff = "noSubThenOff" ;
    public static final String ActutorPort = "actutorPort" ;
    public static final String DbName = "dbName" ;
}

 pipIrr-platform/pipIrr-mw/pipIrr-param-set/src/main/java/com/dy/pipIrrParamSet/paramSet/ParamSetMw.java

@@ -121,6 +121,10 @@
                    newLine = "tcp.port=" + paramMap.get(ParamKey.TcpPort);
                }else if(newLine != null && newLine.trim().startsWith("base.upData.min.interval=")){
                    newLine = "base.upData.min.interval=" + paramMap.get(ParamKey.UpDataMinInterval);
                }else if(newLine != null && newLine.trim().startsWith("mqtt.enable=")){
                    newLine = "mqtt.enable=" + paramMap.get(ParamKey.MqttEnable);
                }else if(newLine != null && newLine.trim().startsWith("mqtt.noSubThenOff=")){
                    newLine = "mqtt.noSubThenOff=" + paramMap.get(ParamKey.MqttNoSubThenOff);
                }
                newLines.add(newLine);
            }
@@ -150,6 +154,12 @@
        key = ParamKey.WebPort;
        this.getConfig(env, map, paramNamePre + key, key);

        key = ParamKey.MqttEnable;
        this.getConfig(env, map, paramNamePre + key, key);

        key = ParamKey.MqttNoSubThenOff;
        this.getConfig(env, map, paramNamePre + key, key);

        key = ParamKey.ActutorPort;
        this.getConfig(env, map, paramNamePre + key, key);


 pipIrr-platform/pipIrr-mw/pipIrr-param-set/src/main/resources/application-params.yml

@@ -4,6 +4,8 @@
            orgTag: ym
            tcpPort: 60000
            upDataMinInterval: 6
            mqttEnable: false
            noSubThenOff: 10
            webPort: 8070
            actutorPort: 9070
            dbName: pipIrr_ym
@@ -11,6 +13,8 @@
            orgTag: sp
            tcpPort: 62000
            upDataMinInterval: 6
            mqttEnable: false
            noSubThenOff: 10
            webPort: 8073
            actutorPort: 9073
            dbName: pipIrr_sp
@@ -18,6 +22,8 @@
            orgTag: test
            tcpPort: 65000
            upDataMinInterval: 6
            mqttEnable: false
            noSubThenOff: 10
            webPort: 8072
            actutorPort: 9072
            dbName: pipIrr_test
@@ -25,6 +31,8 @@
            orgTag: mj
            tcpPort: 61000
            upDataMinInterval: 6
            mqttEnable: false
            noSubThenOff: 10
            webPort: 8071
            actutorPort: 9071
            dbName: pipIrr_mj
@@ -33,6 +41,8 @@
            orgTag: mq
            tcpPort: 60100
            upDataMinInterval: 6
            mqttEnable: true
            noSubThenOff: 10
            webPort: 8100
            actutorPort: 9100
            dbName: pipIrr_mq
@@ -40,6 +50,8 @@
            orgTag: yq
            tcpPort: 60101
            upDataMinInterval: 6
            mqttEnable: false
            noSubThenOff: 10
            webPort: 8101
            actutorPort: 9101
            dbName: pipIrr_yq
@@ -47,6 +59,8 @@
            orgTag: hlj
            tcpPort: 60102
            upDataMinInterval: 6
            mqttEnable: false
            noSubThenOff: 10
            webPort: 8102
            actutorPort: 9102
            dbName: pipIrr_hlj
@@ -54,6 +68,8 @@
            orgTag: gz
            tcpPort: 60103
            upDataMinInterval: 6
            mqttEnable: false
            noSubThenOff: 10
            webPort: 8103
            actutorPort: 9103
            dbName: pipIrr_gz
@@ -61,6 +77,8 @@
            orgTag: lz
            tcpPort: 60104
            upDataMinInterval: 6
            mqttEnable: false
            noSubThenOff: 10
            webPort: 8104
            actutorPort: 9104
            dbName: pipIrr_lz
@@ -68,6 +86,8 @@
            orgTag: jc
            tcpPort: 60105
            upDataMinInterval: 6
            mqttEnable: true
            noSubThenOff: 10
            webPort: 8105
            actutorPort: 9105
            dbName: pipIrr_jc

 pipIrr-platform/Îĵµ/´ò°ü·¢²¼/´ò°ü·¢²¼ÅäÖÃÐÞ¸Ä.docx

Binary files differ