完善代码

liurunyu

2025-08-18 bb25af7205d5f336fd9e48ca72353932ad45abe2

 pipIrr-platform/pipIrr-mw/pipIrr-mwTest-client/src/main/java/com/dy/testClient/tcpClient/TcpClUnit.java

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

import com.dy.common.mw.UnitAdapterInterface;
import com.dy.common.mw.UnitInterface;
import com.dy.common.mw.UnitStartedCallbackInterface;
import com.dy.common.mw.UnitCallbackInterface;
import com.dy.common.threadPool.ThreadPool;
import com.dy.common.threadPool.TreadPoolFactory;
import com.dy.common.util.Callback;
@@ -53,7 +53,7 @@
    }

    @Override
    public void start(UnitStartedCallbackInterface callback) throws Exception {
    public void start(UnitCallbackInterface callback) throws Exception {
        pool = TreadPoolFactory.getThreadPoolLong() ;
        System.out.println("Tcp Client模块成功启动");
        this.doStart();
@@ -61,7 +61,7 @@
    }

    @Override
    public void stop(UnitStartedCallbackInterface callback) throws Exception {
    public void stop(UnitCallbackInterface callback) throws Exception {
        callback.call(null);
    }

@@ -71,7 +71,7 @@
            public void run() {
                try {
                    while(true){
                        if(!ServerProperties.startWork){
                        if(!ServerProperties.startTcpConnectWork){
                            Thread.sleep(100L);
                        }else{
                            try{
@@ -83,10 +83,13 @@
                                log.info("共模拟了" + totalRtuClientCount + "台RTU");

                                Collection<MyThreadJob> collection = jobMap.values() ;
                                int connectedCount = 0 ;
                                for(MyThreadJob job : collection){
                                    connectServer(job) ;
                                    connectedCount++ ;
                                    log.info("当前建立与通信中间件连接的RTU数量为：" + connectedCount);
                                }
                                log.info("启动所有RTU连接通信中间件");
                                log.info("所有RTU已与通信中间件建立连接");

                                while (true){
                                    int noConnectedCount = checkConnected() ;
@@ -98,7 +101,15 @@
                                    }
                                }

                                startJob() ;
                                while (true){
                                    if(!ServerProperties.startRtuReportWork){
                                        Thread.sleep(100L);
                                    }else{
                                        startJob() ;
                                        break ;
                                    }
                                }

                                while(true){
                                    if(totalOverClientCount.longValue() >= totalRtuClientCount.longValue()){
                                        Long seconds = (System.currentTimeMillis() - startTime)/1000 ;
@@ -176,30 +187,27 @@
    }

    private void startJob(){
        new Thread(new Runnable(){
            @Override
            public void run() {
                try {
                    int notOverCount;
                    while(true){
                        notOverCount = 0 ;
                        Collection<MyThreadJob> collection = jobMap.values() ;
                        for(MyThreadJob job : collection){
                            if(!job.isOver){
                                notOverCount++ ;
                                pool.putJob(job);
                            }
                        }
                        if(notOverCount > 0){
                            log.info("当前还有" + notOverCount + "台RTU未完成任务");
                            Thread.sleep(ServerProperties.sendInterval * 1000);
                        }else{
                            break ;
        new Thread(() -> {
            try {
                int notOverCount;
                while(true){
                    notOverCount = 0 ;
                    Collection<MyThreadJob> collection = jobMap.values() ;
                    for(MyThreadJob job : collection){
                        if(!job.isOver){
                            notOverCount++ ;
                            pool.putJob(job);
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    if(notOverCount > 0){
                        log.info("当前还有" + notOverCount + "台RTU未完成任务");
                        Thread.sleep(ServerProperties.sendInterval * 1000);
                    }else{
                        break ;
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();
    }
@@ -209,16 +217,16 @@
        totalSendDataCount++;
        if(totalOverClientCount.longValue() >= totalRtuClientCount.longValue()){
            RmiClUnit.getInstance().reportHadReportCount(totalSendDataCount);
            System.out.println("已经发送" + totalSendDataCount + "条数据");
            System.out.println("已经发送" + totalSendDataCount + "条数据（心跳和上报）");
        }else{
            if(totalRtuClientCount > 100){
                if(totalSendDataCount % 100 == 0){
                    RmiClUnit.getInstance().reportHadReportCount(totalSendDataCount);
                    System.out.println("已经发送" + totalSendDataCount + "条数据");
                    System.out.println("已经发送" + totalSendDataCount + "条数据（心跳和上报）");
                }
            }else{
                RmiClUnit.getInstance().reportHadReportCount(totalSendDataCount);
                System.out.println("已经发送" + totalSendDataCount + "条数据");
                System.out.println("已经发送" + totalSendDataCount + "条数据（心跳和上报）");
            }
        }
    }