1、通信中间件核心模块线程工作由Thread实现改为Timer实现； 2、完善通信中间件远程升级模块及webRemote模块； 3、优化代码。

liurunyu

2024-11-22 1950b0635a0db2c5286f8330e064879e020a7303

1、通信中间件核心模块线程工作由Thread实现改为Timer实现；
2、完善通信中间件远程升级模块及webRemote模块；
3、优化代码。

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/core/CoreConstantManage.java

@@ -1,5 +1,6 @@
package com.dy.common.mw.core;

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

import org.apache.logging.log4j.*;
@@ -10,10 +11,12 @@
    
    private final static CoreConstantManage instance = new CoreConstantManage() ;

    private static Long sleepBigBusy = 100L ;//大忙时（除了恒久任务，还是其他任务），核心线程暂停间隔
    private static Long sleepSmallBusy = 500L ;//小忙时（只有恒久任务，无其他任务），核心线程暂停间隔
	
    private List<CoreConstantTimer> coreConstantTimers ;

    private Long coreInterval = 100L ;//核心线程暂停间隔

    private CoreConstantManage(){
        coreConstantTimers = new ArrayList<>() ;
    }
    
    public static CoreConstantManage getInstance(){
@@ -22,12 +25,10 @@
    
    /**
     * è®¾ç½®æš‚停时长
     * @param sleepBigBusy å¤§å¿™æ—¶ï¼ˆé™¤äº†æ’久任务，还是其他任务），核心线程暂停间隔
     * @param sleepSmallBusy å°å¿™æ—¶ï¼ˆåªæœ‰æ’久任务，无其他任务），核心线程暂停间隔
     * @param coreInterval æ ¸å¿ƒçº¿ç¨‹æš‚停间隔
     */
    public void setSleep(Long sleepBigBusy, Long sleepSmallBusy){
        CoreConstantManage.sleepBigBusy = sleepBigBusy ;
        CoreConstantManage.sleepSmallBusy = sleepSmallBusy ;
    public void setSleep(Long coreInterval){
        this.coreInterval = coreInterval ;
    }

    public void start(){
@@ -35,7 +36,8 @@
        List<CoreTask> constantTasks = CoreUnit.getAllConstantTasks();
        if (constantTasks != null && constantTasks.size() > 0) {
            for (CoreTask task : constantTasks) {
                new CoreConstantThread(sleepBigBusy, sleepSmallBusy, task).start();
                //new CoreConstantThread(sleepBigBusy, sleepSmallBusy, task).start();
                coreConstantTimers.add(new CoreConstantTimer(coreInterval, task).start());
            }
        }
    }

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/core/CoreConstantThread.java

@@ -9,6 +9,15 @@
 * @LastEditTime 2023/12/19 16:41
 * @Description
 */
/**
 * 2024-11-21 è¯¥ç±»å·²ç»å¼ƒç”¨ï¼Œæ”¹ç”±CoreConstantTimer实现。
 * å½“采用Thread实现时，在while(true){}中使用Thread.sleep()，可能导致的严重性能问题，不推荐在循环中使用线程休眠。
 * Java线程实现采用内核线程实现，线程的休眠及唤醒（状态切换）需借助操作系统进行，这是一个极其耗时耗力的操作。
 * åœ¨çº¿ç¨‹ä¼‘眠或运行时间较长的情景下，其对性能的影响还不算明显，因为对线程状态的切换并不频繁。
 * ä½†è‹¥çº¿ç¨‹ä¼‘眠及运行的时间都很短（例如毫秒/秒），
 * ç³»ç»Ÿå°†é¢‘繁的对线程状态进行切换，导致严重的性能损耗，并对着循环次数的递增而放大。
 */
@Deprecated
public class CoreConstantThread extends Thread {

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

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/core/CoreConstantTimer.java

New file
@@ -0,0 +1,45 @@
package com.dy.common.mw.core;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.Timer;
import java.util.TimerTask;

/**
 * @Author: liurunyu
 * @Date: 2024/11/21 17:13
 * @Description
 */
public class CoreConstantTimer extends TimerTask {

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

    private Timer timer;

    private long coreInterval ;

    private CoreTask task ;

    public CoreConstantTimer(long coreInterval, CoreTask task){
        this.timer = new Timer();
        this.coreInterval = coreInterval ;
        this.task = task ;
    }

    public CoreConstantTimer start(){
        this.timer.schedule(this, 0 , this.coreInterval);
        return this ;
    }

    @Override
    public void run() {
        if(task != null){
            try {
                task.execute();
            } catch (Exception e) {
                log.error("恒久任务" + task.getClass().getName() + "执行时发生异常" + (e.getMessage() == null ? "" : (":" + e.getMessage())), e);
            }
        }
    }
}

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/core/CoreThread.java

@@ -4,6 +4,15 @@

import com.dy.common.queue.Queue;

/**
 * 2024-11-21 è¯¥ç±»å·²ç»å¼ƒç”¨ï¼Œæ”¹ç”±CoreTimer实现。
 * å½“采用Thread实现时，在while(true){}中使用Thread.sleep()，可能导致的严重性能问题，不推荐在循环中使用线程休眠。
 * Java线程实现采用内核线程实现，线程的休眠及唤醒（状态切换）需借助操作系统进行，这是一个极其耗时耗力的操作。
 * åœ¨çº¿ç¨‹ä¼‘眠或运行时间较长的情景下，其对性能的影响还不算明显，因为对线程状态的切换并不频繁。
 * ä½†è‹¥çº¿ç¨‹ä¼‘眠及运行的时间都很短（例如毫秒/秒），
 * ç³»ç»Ÿå°†é¢‘繁的对线程状态进行切换，导致严重的性能损耗，并对着循环次数的递增而放大。
 */
@Deprecated
public class CoreThread extends Thread{
    
    private static Logger log = LogManager.getLogger(CoreThread.class.getName()) ;

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/core/CoreTimer.java

New file
@@ -0,0 +1,80 @@
package com.dy.common.mw.core;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.Timer;
import java.util.TimerTask;

/**
 * @Author: liurunyu
 * @Date: 2024/11/21 13:41
 * @Description
 */
public class CoreTimer extends TimerTask {

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

    private static final CoreTimer instance = new CoreTimer() ;

    private Long workInterval = 100L ;//核心线程暂停间隔

    private Timer timer;

    private boolean stop;

    private CoreTimer(){
        this.timer = new Timer();
        this.stop = false ;
    }

    public static CoreTimer getInstance(){
        return instance ;
    }

    public void stop(){
        this.stop = true ;
        if(this.timer != null){
            this.timer.cancel();
        }
    }
    public boolean isStop(){
        return this.stop ;
    }

    /**
     * è®¾ç½®/核心线程暂停间隔
     * @param workInterval /核心线程暂停间隔
     */
    public void setSleep(Long workInterval){
        this.workInterval = workInterval ;
    }

    public void start(){
        this.timer.schedule(this, 0 , this.workInterval);
    }

    /**
     * TimerTask的run方法
     */
    @Override
    public void run() {
        try{
            /**
             * æ­¤å¤„取出当前队列的节点数量，作为一个定量count处理，
             * åŽŸå› :
             * åœ¨å¤„理定量count过程中，可能有的节点又回到队列中，也可能新生节点进入队列中，即在处理过程中，队列可能又增长了
             */
            int count = CoreUnit.taskQueue.size() ;
            while(count > 0){
                CoreTask task = (CoreTask)CoreUnit.taskQueue.pop() ;
                if(task != null){
                    task.execute();
                }
                count-- ;
            }
        }catch(Exception e){
            log.error("核心线程发生异常" + (e.getMessage() == null ? "" : (":" + e.getMessage())), e);;
        }
    }
}

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/core/CoreUnit.java

@@ -8,6 +8,7 @@
import com.dy.common.mw.UnitInterface;
import com.dy.common.mw.UnitCallbackInterface;


public class CoreUnit implements UnitInterface {
    
    private static final CoreUnit instance = new CoreUnit() ;
@@ -36,17 +37,14 @@
        if(vo == null){
            throw new Exception("核心模块配置对象不能为空！") ;
        }
        if(vo.sleepBigBusy == null || vo.sleepSmallBusy == null){
        if(vo.coreInterval == null){
            throw new Exception("核心模块配置对象间隔属性值不能为空！") ;
        }
        if(vo.sleepBigBusy <= 0){
            throw new Exception("核心模块配置对象属性sleepBigBusy值不能小于0！") ;
        if(vo.coreInterval <= 0){
            throw new Exception("核心模块配置对象属性coreInterval值不能小于0！") ;
        }
        if(vo.sleepBigBusy > 1000){
            throw new Exception("核心模块配置对象属性sleepBigBusy值不能大于1000！") ;
        }
        if(vo.sleepSmallBusy > 1000){
            throw new Exception("核心模块配置对象属性sleepSmallBusy值不能大于1000！") ;
        if(vo.coreInterval > 1000){
            throw new Exception("核心模块配置对象属性coreInterval值不能大于1000！") ;
        }
        if(vo.queueWarnSize == null || vo.queueMaxSize == null){
            throw new Exception("核心模块配置对象队列节点限制数量属性值不能为空！") ;
@@ -67,18 +65,22 @@
    public void start(UnitCallbackInterface callback) throws Exception {
        if(!started){
            started = true ;
            /*
            CoreThread ct = CoreThread.getInstance() ;
            ct.setSleep(this.adapter.getConfig().sleepBigBusy, this.adapter.getConfig().sleepSmallBusy);
            ct.start(); 
			
            ct.start();
            */
            CoreTimer ct = CoreTimer.getInstance() ;
            ct.setSleep(this.adapter.getConfig().coreInterval);
            ct.start();

            CoreConstantManage ccm = CoreConstantManage.getInstance() ;
            ccm.setSleep(this.adapter.getConfig().sleepBigBusy, this.adapter.getConfig().sleepSmallBusy);
            ccm.setSleep(this.adapter.getConfig().coreInterval);
            ccm.start();
            
            if(adapter.getConfig().showStartInfo){
                System.out.println("核心模块成功启动，"
                        + "主线程繁忙间隔：" + adapter.getConfig().sleepBigBusy + "毫秒，"
                        + "轻闲间隔：" + adapter.getConfig().sleepSmallBusy + "毫秒"  );
                        + "主线程执行间隔：" + adapter.getConfig().coreInterval + "毫秒");
            }
            callback.call(null);
        }

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/mw/core/CoreUnitConfigVo.java

@@ -1,8 +1,7 @@
package com.dy.common.mw.core;

public class CoreUnitConfigVo {
    public Long sleepBigBusy = 100L ;//大忙时（除了恒久任务，还有其他任务需处理），核心线程暂停间隔
    public Long sleepSmallBusy = 500L ;//小忙时（只有恒久任务，无其他任务需处理），核心线程暂停间隔
    public Long coreInterval = 100L ;//大忙时（除了恒久任务，还有其他任务需处理），核心线程暂停间隔
    public Integer queueWarnSize = 1000 ;//队列报警长度
    public Integer queueMaxSize = 5000 ;//队列最大长度
    public Boolean showStartInfo; //是否在控制台上显示信息

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/util/ThreadJob.java

@@ -2,6 +2,7 @@


@SuppressWarnings("unused")
@Deprecated
public abstract class ThreadJob {
    protected Thread thread ;
    protected boolean stop = false ;

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/util/TimerTaskJob.java

@@ -7,11 +7,12 @@
 * @Author: liurunyu
 * @Date: 2024/11/20 17:49
 * @Description :
 * 2024-11-20：
 * æ­¤æŠ½è±¡ç±»ä»£æ›¿ThreadJob，原因：
 * åœ¨while(true){}中使用Thread.sleep()，可能导致的严重性能问题，不推荐在循环中使用线程休眠。
 * Java线程实现采用内核线程实现，线程的休眠及唤醒（状态切换）需借助操作系统进行，这是一个极其耗时耗力的操作。
 * åœ¨çº¿ç¨‹ä¼‘眠或运行时间较长的情景下，其对性能的影响还不算明显，因为对线程状态的切换并不频繁。
 * ä½†è‹¥çº¿ç¨‹ä¼‘眠及运行的时间都很短（例如毫秒/秒，文中案例就是一个典型案例），
 * ä½†è‹¥çº¿ç¨‹ä¼‘眠及运行的时间都很短（例如毫秒/秒），
 * ç³»ç»Ÿå°†é¢‘繁的对线程状态进行切换，导致严重的性能损耗，并对着循环次数的递增而放大。
 */
public abstract class TimerTaskJob {

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

@@ -370,8 +370,7 @@
            // ///////////////
            // æ ¸å¿ƒ
            CoreUnitConfigVo coreConfVo = new CoreUnitConfigVo();
            coreConfVo.sleepBigBusy = conf.getSetAttrPlusInt(doc, "config.core", "sleepBigBusy", null, 1, 200, null).longValue() ;
            coreConfVo.sleepSmallBusy = conf.getSetAttrPlusInt(doc, "config.core", "sleepSmallBusy", null, 2, 1000, null).longValue();
            coreConfVo.coreInterval = conf.getSetAttrPlusInt(doc, "config.core", "coreInterval", null, 1, 200, null).longValue() ;
            coreConfVo.queueWarnSize = ServerProperties.cacheUpDownDataWarnCount ;
            coreConfVo.queueMaxSize = ServerProperties.cacheUpDownDataMaxCount ;
            coreConfVo.showStartInfo = showStartInfo ;

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

@@ -13,6 +13,7 @@

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

/**
 * @Author: liurunyu
@@ -202,6 +203,7 @@
        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().stream().forEach(info ->{
                    if(info.state == UpgradeRtu.STATE_OFFLINE){
                        state.offLineTotal ++ ;
@@ -222,8 +224,15 @@
                    }
                    if(info.isOver){
                        state.overTotal++;
                    }else{
                        hasRunning.set(true);
                    }
                });
                if(!hasRunning.get()){
                    state.allOver = true ;
                }else{
                    state.allOver = false ;
                }
            }
        }
        return state ;

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/resources/config(spɳÅÌ).xml

@@ -124,14 +124,12 @@
    />

    <!--
    sleepBigBusy：主线程大忙时每次执行间隔时长（毫秒）
    sleepSmallBusy：主线程小忙时每次执行间隔时长（毫秒）
    coreInterval：主线程每次执行间隔时长（毫秒）
    queueWarnSize: ç¼“存队列节点数警告数量
    queueMaxSize: ç¼“存队列节点数最大数量
     -->
    <core
            sleepBigBusy="100"
            sleepSmallBusy="500"
            coreInterval="100"
            queueWarnSize="100000"
            queueMaxSize="300000"
    />

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/resources/config(test²âÊÔ).xml

@@ -124,14 +124,12 @@
    />

    <!--
    sleepBigBusy：主线程大忙时每次执行间隔时长（毫秒）
    sleepSmallBusy：主线程小忙时每次执行间隔时长（毫秒）
    coreInterval：主线程每次执行间隔时长（毫秒）
    queueWarnSize: ç¼“存队列节点数警告数量
    queueMaxSize: ç¼“存队列节点数最大数量
     -->
    <core
            sleepBigBusy="100"
            sleepSmallBusy="500"
            coreInterval="100"
            queueWarnSize="100000"
            queueMaxSize="300000"
    />

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/resources/config(ymԪı).xml

@@ -124,14 +124,12 @@
    />

    <!--
    sleepBigBusy：主线程大忙时每次执行间隔时长（毫秒）
    sleepSmallBusy：主线程小忙时每次执行间隔时长（毫秒）
    coreInterval：主线程每次执行间隔时长（毫秒）
    queueWarnSize: ç¼“存队列节点数警告数量
    queueMaxSize: ç¼“存队列节点数最大数量
     -->
    <core
            sleepBigBusy="100"
            sleepSmallBusy="500"
            coreInterval="100"
            queueWarnSize="100000"
            queueMaxSize="300000"
    />

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

@@ -124,14 +124,12 @@
    />

    <!--
    sleepBigBusy：主线程大忙时每次执行间隔时长（毫秒）
    sleepSmallBusy：主线程小忙时每次执行间隔时长（毫秒）
    coreInterval：主线程每次执行间隔时长（毫秒）
    queueWarnSize: ç¼“存队列节点数警告数量
    queueMaxSize: ç¼“存队列节点数最大数量
     -->
    <core
            sleepBigBusy="100"
            sleepSmallBusy="500"
            coreInterval="100"
            queueWarnSize="100000"
            queueMaxSize="300000"
    />

 pipIrr-platform/pipIrr-web/pipIrr-web-remote/src/main/java/com/dy/pipIrrRemote/common/CommandSv.java

@@ -132,7 +132,7 @@
                    jsonArray.add(jsonObject);
                });
            }
            System.out.println(jsonArray);
            //System.out.println(jsonArray);
        }

        /**

 pipIrr-platform/pipIrr-web/pipIrr-web-remote/src/main/java/com/dy/pipIrrRemote/rtuUpgrage/RtuUpgradeStateReceiverCtrl.java

@@ -244,7 +244,7 @@
        }

        int n = Integer.parseInt(new CreateRandom().create(3)) ;
        if(n == 540 || n == 541 || n == 542 || n == 543 || n == 544 || n == 545){
        if(n == 540 || n == 541 || n == 542 || n == 543 || n == 544 || n == 545 || n == 546 || n == 547 || n == 548 || n == 549 ){
            if(rtu.currentPackage == 1){
                //1包死
                rtu.state = UpgradeRtu.STATE_FAILONE ;