采用模拟软件MQTTX进行模拟测试，发现并修改MQTT相关水肥机上行数据与下行命令逻辑bug.

liurunyu

2025-06-23 431bbfaddba9f39496359bc0c1e2dcaf2bb6fca2

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/server/upgrade/UpgradeTask.java

@@ -1,13 +1,21 @@
package com.dy.rtuMw.server.upgrade;

import com.alibaba.fastjson2.annotation.JSONField;
import com.dy.common.mw.protocol.rtuState.RtuStatus;
import com.dy.common.softUpgrade.parse.HexFileParse;
import com.dy.common.softUpgrade.parse.HexFileVo;
import com.dy.common.softUpgrade.state.UpgradeRtu;
import com.dy.common.softUpgrade.state.UpgradeState;
import com.dy.common.softUpgrade.state.UpgradeTaskVo;
import com.dy.common.util.Callback;
import com.dy.common.util.DateTime;
import com.dy.rtuMw.server.forTcp.RtuStatusDealer;
import lombok.Data;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @Author: liurunyu
@@ -17,136 +25,200 @@
@Data
public class UpgradeTask {

    private static final Logger log = LogManager.getLogger(UpgradeTask.class.getName());

    protected static final String TaskOverType_Natural = "自然" ;
    protected static final String TaskOverType_Force = "强制" ;

    @JSONField(serialize = false)
    protected Integer failTryTimes ;//升级失败后，重新偿试升级次数，0表示不重新偿试升级
    @JSONField(serialize = false)
    protected Integer ugMaxRtuSameTime ;//同时升级RTU最大个数
    @JSONField(serialize = false)
    private Boolean openNoUpgrade ;//阀开（泵开）不执行升级
    @JSONField(serialize = false)
    private Integer lastOpenMaxGoOn ;//阀开（泵开）状态设置以来持续最长时间（秒钟），超过这个时间认为状态无效（这个时长取决于工作报间隔）

    public String setupDt ;//设置时间(yyyy-mm-dd HH:MM:SS)
    @JSONField(serialize = false)
    private Long setupDtLong ;//设置时间

    public String softFileName ;//升级软件（hex）文件名称
    public String softStoreAddr ;//升级程序存放地址(4字节，8字符HEX字符串)，升级程序在FLASH中存放地址
    public String softStartAddr ;//程序覆盖起始地址(4字节，8字符HEX字符串)，被刷新程序的起始地址高字节在前 ，低字节在后
    public UpgradeTaskVo taskVo ;//升级任务值对象

    @JSONField(serialize = false)
    public byte[][] softData ;//升级程序数据(每包数据是512字节)
    @JSONField(serialize = false)
    public int softByteSrc16;//升级程序校验码 CRC16

    public int softBytesCalculate;//升级程序字节数(按公式计算)

    public List<String> rtuAddrList ;//需要升级的RTU地址集合
    protected byte[][] softFileDataGrp ;//以512字节为单位把升级程序数据分组

    @JSONField(serialize = false)
    public Map<String, UpgradeRtu> upgradeState ;//升级状态
    protected ConcurrentHashMap<String, UpgradeRtu> upgradeRtus;//升级状态

    public boolean taskIsOver = false ;//任务是否完成
    public String taskOverType = "" ;//任务完成方式（自然，强制）
    public String taskOverDt = "" ;//任务完成时间(yyyy-mm-dd HH:MM:SS)

    ///////////////////////////////////////////////////
    //以下内部控制用
    @JSONField(serialize = false)
    protected int curUgRunningRtuTotal = 0 ;//当前正在升级的RTU个数

    public UpgradeTask() {
        this.curUgRunningRtuTotal = 0 ;
    }
    /**
     *  初始化配置信息
     */
    public void initOption(Integer failTryTimes, Integer ugMaxRtuSameTime) {
    public void initOption(Boolean openNoUpgrade,
                           Integer lastOpenMaxGoOn,
                           Integer failTryTimes,
                           Integer ugMaxRtuSameTime) {
        this.openNoUpgrade = openNoUpgrade;
        this.lastOpenMaxGoOn = lastOpenMaxGoOn;
        this.failTryTimes = failTryTimes;
        this.ugMaxRtuSameTime = ugMaxRtuSameTime;
    }
    /**
     * 设置升级任务
     * @param softFileName 升级程序文件名
     * @param softStoreAddr 升级程序存放地址
     * @param softStartAddr 程序覆盖起始地址
     * @param softFileData 升级程序字节数组
     * @param softBytesCalculate 升级程序字节数(按公式计算)
     * @param rtuAddrList 升级RTU
     * @param taskVo UpgradeTaskVo 升级任务对象
     * @throws Exception
     */
    public void setTask(String softFileName,
                       String softStoreAddr,
                       String softStartAddr,
                       byte[] softFileData,
                       Integer softBytesCalculate,
                       List<String> rtuAddrList) throws Exception {
        if(softFileName == null || softFileName.trim().length() == 0){
    public void setTask(UpgradeTaskVo taskVo) throws Exception {
        if(taskVo.id == null || taskVo.id.trim().length() == 0){
            throw new Exception("升级任务id必须提供") ;
        }
       if(taskVo.softFileName == null || taskVo.softFileName.trim().length() == 0){
            throw new Exception("升级软件（hex）文件名称必须提供") ;
        }
        if(softStoreAddr == null || softStoreAddr.trim().length() != 8){
        if(taskVo.softStoreAddr == null || taskVo.softStoreAddr.trim().length() != 8){
            throw new Exception("升级程序存放地址不合法，必须是8字符（十六进制）的字符串") ;
        }
        if(softStartAddr == null || softStartAddr.trim().length() != 8){
        if(taskVo.softStartAddr == null || taskVo.softStartAddr.trim().length() != 8){
            throw new Exception("程序覆盖起始地址不合法，必须是8字符（十六进制）的字符串") ;
        }
        if(softFileData == null || softFileData.length <= 0){
        if(taskVo.softFileData64 == null || taskVo.softFileData64.trim().length() == 0){
            throw new Exception("升级程序内容必须提供") ;
        }
        if(rtuAddrList == null || rtuAddrList.size() <= 0){
        if(taskVo.softBytesCalculate == null){
            throw new Exception("公式计算升级程序有效序字节数必须提供") ;
        }
       if(taskVo.softByteSrc16 == null){
            throw new Exception("有效升级程序字节数CRC16校验值必须提供") ;
        }
        if(taskVo.rtuAddrList == null || taskVo.rtuAddrList.size() <= 0){
            throw new Exception("升级设备RTU地址必须提供") ;
        }
        if(taskVo.callbackWebUrl == null || taskVo.callbackWebUrl.trim().equals("")){
            throw new Exception("回调网址必须提供") ;
        }
        this.setupDt = DateTime.yyyy_MM_dd_HH_mm_ss() ;
        this.softFileName = softFileName;
        this.softStoreAddr = softStoreAddr;
        this.softStartAddr = softStartAddr;
        this.softBytesCalculate = softBytesCalculate;
        this.rtuAddrList = rtuAddrList;
        this.setupDtLong = System.currentTimeMillis() ;
        this.taskVo = taskVo ;

        this.upgradeState = new HashMap<>();
        if(softFileData != null && softFileData.length >0){
            HexFileVo vo = new HexFileParse().doParse(softFileData);
            this.softData = vo.listByte512.toArray(new byte[0][]);
            this.softByteSrc16 = vo.bytesCrc16 ;
        this.upgradeRtus = new ConcurrentHashMap<>();
        if(taskVo.softFileData64 != null && !taskVo.softFileData64.trim().equals("")){
            taskVo.softFileData = Base64.getDecoder().decode(taskVo.softFileData64);
            List<byte[]> listBytes = new HexFileParse().splitBytesByUnit512(taskVo.softFileData);
            this.softFileDataGrp = listBytes.toArray(new byte[0][]);
            try{
                for(String rtuAddr : this.taskVo.rtuAddrList){
                    //此时状态设置成离线状态
                    UpgradeRtuDev ugRtu = new UpgradeRtuDev(this.taskVo, this.failTryTimes, rtuAddr, this.softFileDataGrp.length, UpgradeRtuDev.STATE_OFFLINE) ;
                    this.upgradeRtus.put(rtuAddr, ugRtu) ;
                }
            }catch (Exception e){
                log.error(e);
            }
        }
    }
    /**
     * RTU有上行数据了，触发下发升级数据
     * @param rtuAddr
     * @param code
     * @param callback
     * @param callbackCom
     */
    public void trigger(String rtuAddr, String code, String protocolName, Short protocolVersion, Callback callback){
        if(upgradeState != null && upgradeState.size() > 0
                && rtuAddrList != null && rtuAddrList.size() > 0){
            UpgradeRtu info = upgradeState.get(rtuAddr) ;
            if(info == null){
                if(rtuAddrList.contains(rtuAddr)){
                    info = new UpgradeRtu(this, rtuAddr, softData.length) ;
                    upgradeState.put(rtuAddr, info) ;
    public void trigger(String rtuAddr, String code, String protocolName, Short protocolVersion, Callback callbackCom, Object ...objects){
        if(this.upgradeRtus != null && this.upgradeRtus.size() > 0
                && this.taskVo.rtuAddrList != null && this.taskVo.rtuAddrList.size() > 0){
            UpgradeRtu ugRtu = this.upgradeRtus.get(rtuAddr) ;
            if(ugRtu == null){
                //根据方法setTask的逻辑，只要RTU在升级之列，此处ugRtuState一定不为null
                //为保险，实现下面逻辑
                if(taskVo.rtuAddrList.contains(rtuAddr)){
                    ugRtu = new UpgradeRtuDev(this.taskVo, this.failTryTimes, rtuAddr, this.taskVo.softFileData.length) ;
                    upgradeRtus.put(rtuAddr, ugRtu) ;
                }else{
                    //rtu不在升级之列
                    return ;
                }
            }
            if(info != null){
                info.trigger(code, protocolName, protocolVersion, this.softData, callback) ;
            }else{
                if(ugRtu.isOver){
                    //当前RTU已经升级完成，无需再升级
                    return;
                }else{
                    if(this.openNoUpgrade){
                        //首先判断是否是阀门打开状态
                        RtuStatus rtuStatus = RtuStatusDealer.oneStatus(rtuAddr) ;
                        if(rtuStatus != null
                                && rtuStatus.valveOpenTrueCloseFalse != null
                                && rtuStatus.valveOpenTrueCloseFalse.booleanValue() == true
                                && rtuStatus.valveStatusLastTimeStamp != null){
                            //有状态，并且是阀开（泵开）
                            Long now = System.currentTimeMillis() ;
                            Long gap = now - rtuStatus.valveStatusLastTimeStamp ;
                            if(gap < this.lastOpenMaxGoOn){
                                //这时采纳阀门打开状态，进而不能升级
                                ugRtu.isOver = true ;
                                ugRtu.state = UpgradeRtu.STATE_FAILOPEN ;
                                return;
                            }
                        }
                    }

                    if(UpgradeUnit.confVo.ugMaxRtuAtOnce > 0){
                        //设置了同时升级的RTU最大数量的限制
                        if(ugRtu.state == UpgradeRtuDev.STATE_OFFLINE){
                            //初始态，说明升级任务设置以来，该RTU第一次上行数据
                            if(this.curUgRunningRtuTotal <= UpgradeUnit.confVo.ugMaxRtuAtOnce){
                                //当前正在升级的RTU数量还未受限
                                ugRtu.trigger(code, protocolName, protocolVersion, this.softFileDataGrp, callbackCom, objects) ;
                            }else{
                                //同时升级的RTU数量受限，等待下次机会
                                //但先表明一下状态
                                ugRtu.state = UpgradeRtuDev.STATE_UNSTART ;
                                return ;
                            }
                        }else if(ugRtu.state == UpgradeRtuDev.STATE_UNSTART){
                            //根据上面逻辑， 说明必然在线了
                            if(this.curUgRunningRtuTotal <= UpgradeUnit.confVo.ugMaxRtuAtOnce){
                                //当前正在升级的RTU数量还未受限
                                ugRtu.trigger(code, protocolName, protocolVersion, this.softFileDataGrp, callbackCom, objects) ;
                            }else{
                                //同时升级的RTU数量受限，等待下次机会
                                return ;
                            }
                        }else{
                            //RTU已经处于升级过程中
                            ugRtu.trigger(code, protocolName, protocolVersion, this.softFileDataGrp, callbackCom, objects) ;
                        }
                    }else{
                        //没有设置同时升级的RTU最大数量的限制
                         ugRtu.trigger(code, protocolName, protocolVersion, this.softFileDataGrp, callbackCom, objects) ;
                    }
                }
            }
        }
    }

    /**
     * 强制结束升级任务
     */
    public void forceOver(){
        this.rtuAddrList.clear();
        this.upgradeState.clear();
        if(!this.taskIsOver){
            this.taskIsOver = true ;//强制设置任务完成
            this.taskOverType = TaskOverType_Force ;//任务完成方式（自然，强制）
            this.taskOverDt = DateTime.yyyy_MM_dd_HH_mm_ss() ;//任务完成时间(yyyy-mm-dd HH:MM:SS)
            //this.taskVo.rtuAddrList.clear();
            //this.upgradeState.clear();
        }
    }

    /**
     * 升级任务是否完成
     * @return
     */
    public boolean isOver() {
        boolean isOver = true ;
        if(upgradeState != null && upgradeState.size() > 0){
            Collection<UpgradeRtu> col = upgradeState.values() ;
            for(UpgradeRtu info : col){
                if(info.state == UpgradeRtu.STATE_UNSTART){
                    isOver = false ;
                    break ;
                }else if(info.state == UpgradeRtu.STATE_RUNNING){
                    isOver = false ;
                    break ;
                }
            }
        }
        return isOver ;
    }

    /**
     * 当前升级状态
@@ -154,26 +226,46 @@
     */
    public UpgradeState currentUpgradeState() {
        UpgradeState state = new UpgradeState() ;
        if(rtuAddrList != null && rtuAddrList.size() > 0){
            state.rtuTotal = rtuAddrList.size() ;
            if(upgradeState != null && upgradeState.size() > 0){
                Collection<UpgradeRtu> col = upgradeState.values() ;
                for(UpgradeRtu info : col){
                    if(info.state == UpgradeRtu.STATE_UNSTART){
        if(this.taskVo.rtuAddrList != null && this.taskVo.rtuAddrList.size() > 0){
            state.rtuTotal = this.taskVo.rtuAddrList.size() ;
            if(this.upgradeRtus != null && this.upgradeRtus.size() > 0){
                AtomicBoolean hasRunning = new AtomicBoolean(false);
                this.upgradeRtus.values().forEach(info ->{
                    if(info.state == UpgradeRtu.STATE_OFFLINE){
                        state.offLineTotal++ ;
                    }else if(info.state == UpgradeRtu.STATE_UNSTART){
                        state.unStartTotal ++ ;
                    }else if(info.state == UpgradeRtu.STATE_RUNNING){
                        state.runningTotal ++ ;
                    }else if(info.state == UpgradeRtu.STATE_SUCCESS) {
                        state.successTotal++;
                        state.overTotal++;
                    }else if(info.state == UpgradeRtu.STATE_FAILONE) {
                        state.failOneTotal++;
                        state.failTotal++;
                        state.overTotal++;
                        state.dieOneTotal++;
                        if(info.isOver){
                            state.failTotal++;
                        }
                    }else if(info.state == UpgradeRtu.STATE_FAIL) {
                        state.dieMultiTotal++;
                        if(info.isOver) {
                            state.failTotal++;
                        }
                    }else if(info.state == UpgradeRtu.STATE_FAILOFFLINE) {
                        state.failTotal++;
                        state.overTotal++;
                        state.failOffTotal++;
                    }else if(info.state == UpgradeRtu.STATE_FAILOPEN) {
                        state.failTotal++;
                        state.failOpenTotal++;
                    }
                    if(info.isOver){
                        state.overTotal++;
                    }else{
                        hasRunning.set(true);
                    }
                });
                if(!hasRunning.get()){
                    state.allOver = true ;
                }else{
                    state.allOver = false ;
                }
            }
        }
@@ -186,23 +278,174 @@
     * @return
    */
    public UpgradeRtu upgradeInfos(String rtuAddr){
        return upgradeState.get(rtuAddr) ;
        return this.upgradeRtus.get(rtuAddr) ;
    }

    /**
     * Rtu升级信息
     * 一些Rtu升级信息
     * @param rtuAddrList
     * @return
     */
    public List<UpgradeRtu> upgradeInfos(List<String> rtuAddrList){
        List<UpgradeRtu> list = new ArrayList<>() ;
        for(String rtuAddr : rtuAddrList){
            UpgradeRtu info = upgradeState.get(rtuAddr) ;
            UpgradeRtu info = this.upgradeRtus.get(rtuAddr) ;
            if(info != null){
                list.add(info) ;
            }
        }
        return list ;
    }
    /**
     * 全部Rtu升级信息
     * @return
     */
    public List<UpgradeRtu> upgradeInfoAll(){
        if(this.upgradeRtus != null && this.upgradeRtus.size() > 0){
            return this.upgradeRtus.values().stream().toList();
        }else{
            return null ;
        }
    }

    ///////////////////////////////////////////////////////////
    //
    //   以下方法为内部服务，不对外提供服务
    //
    ///////////////////////////////////////////////////////////

    /**
     * 阀开（泵开）不升级处理
     */
    protected void openNoUpgrade(){
        if(this.upgradeRtus != null && this.upgradeRtus.size() > 0){
            Map<String, RtuStatus> rsAllMap = RtuStatusDealer.allStatus() ;
            Long now = System.currentTimeMillis() ;
            this.upgradeRtus.values().stream().forEach(a -> {
                RtuStatus rs = rsAllMap.get(a.rtuAddr) ;
                if(rs != null
                        && rs.valveOpenTrueCloseFalse != null && rs.valveOpenTrueCloseFalse.booleanValue() == true
                        && rs.valveStatusLastTimeStamp != null){
                    //有状态，并且是阀开（泵开）
                    Long gap = now - rs.valveStatusLastTimeStamp ;
                    if(gap < this.lastOpenMaxGoOn){
                        //这时采纳阀门打开状态不能升级
                        a.isOver = true ;
                        a.state = UpgradeRtu.STATE_FAILOPEN ;
                    }
                }
            });
        }
    }
    /**
     * 判断是否没用任何一个RTU进行过升级，而且超过了时限
     * @return -1:无一RTU升级且超时，0：无RTU升级但未超时等待，1有RTU升级正常执行
     */
    protected int countNoOneRtuUpgradeInDuration(){
        if(this.upgradeRtus == null || upgradeRtus.size() == 0){
            //当前没有任何一个设备进行过升级
            Long now = System.currentTimeMillis() ;
            if(now - this.setupDtLong > UpgradeUnit.confVo.noOneRtuUpgradeMaxDuration){
                return -1 ;
            }
        }else{
            Collection<UpgradeRtu> col = this.upgradeRtus.values() ;
            for(UpgradeRtu info : col){
                if(info.currentPackage > 0){
                    //当前有设备进行过升级
                    return 1 ;
                }
            }
            Long now = System.currentTimeMillis() ;
            if(now - this.setupDtLong > UpgradeUnit.confVo.noOneRtuUpgradeMaxDuration){
                return -1 ;
            }
        }
        return 0 ;
    }

    /**
     * 统计当前正在升级的RTU数量，为同时升级数量限制做准备
     */
    protected int countRunningRtuCount(){
        int runningTotal = 0 ;
        Collection<UpgradeRtu> col = this.upgradeRtus.values() ;
        for(UpgradeRtu info : col){
            if(info.state == UpgradeRtu.STATE_RUNNING){
                runningTotal ++ ;
            }
        }
        return this.curUgRunningRtuTotal = runningTotal ;
    }

    /**
     * 统计需要升级但当前离线RTU的情况，超过时限的设备为升级完成
     * @return -1:没有超时，0超时了且无因离线被强制设置升级完成的RTU，>0离线被强制设置升级完成的RTU数量
     */
    protected int countOffRtuAndSetIfOver() {
        Long now = System.currentTimeMillis() ;
        if(now - this.setupDtLong > UpgradeUnit.confVo.rtuOffLineWaitDuration){
            //rtu离线，等待其升级的时长(毫秒)，超过配置的最大时长，设置其升级失败，且设置升级任务完成
            int count = 0 ;
            if (this.taskVo.rtuAddrList != null && this.taskVo.rtuAddrList.size() > 0) {
                Collection<UpgradeRtu> col = this.upgradeRtus.values() ;
                for(UpgradeRtu info : col){
                    if(info.state == UpgradeRtu.STATE_OFFLINE){
                        info.isOver = true ;
                        info.state = UpgradeRtu.STATE_FAILOFFLINE ;
                        count ++ ;
                    }
                }
            }
            return count ;
        }else{
            return -1 ;
        }
    }

    /**
     * 统计:已经进升级但RTU又进入停止升级发呆状态，超过一定时限，设置设备一包死或多包死，并设置为升级完成
     * @return -1:没有超时，0超时了且无因离线被强制设置升级完成的RTU，>0离线被强制设置升级完成的RTU数量
     */
    protected int countRunningIdleRtuAndSetIfOver() {
        Long now = System.currentTimeMillis() ;
        int count = -1 ;
        if(now - this.setupDtLong > UpgradeUnit.confVo.rtuOffLineWaitDuration){
            //设置上句，防止频繁进入下面语句进行计算
            if (this.taskVo.rtuAddrList != null && this.taskVo.rtuAddrList.size() > 0) {
                Collection<UpgradeRtu> col = this.upgradeRtus.values() ;
                for(UpgradeRtu info : col){
                    if(info.state == UpgradeRtu.STATE_RUNNING && info.isOver == false){
                        //升级中，但未升级完成
                        if(now - info.lastDownDtAt > UpgradeUnit.confVo.runningAndIdleDuration){
                            if(info.currentPackage <= 1){
                                //一包死
                                info.state = UpgradeRtu.STATE_FAILONE ;
                            }else{
                                //多包死
                                info.state = UpgradeRtu.STATE_FAIL ;
                            }
                            info.isOver = true ;
                            count ++ ;
                        }
                    }
                }
            }
        }
        return count ;
    }
    /**
     * 统计是否升级全部结束
     */
    protected boolean countIsAllOver() {
        if (this.taskVo.rtuAddrList != null && this.taskVo.rtuAddrList.size() > 0) {
            Collection<UpgradeRtu> col = this.upgradeRtus.values() ;
            for(UpgradeRtu info : col){
                if(info.isOver == false){
                    return false ;
                }
            }
        }
        return true ;
    }
}