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

wuzeyu

2023-12-02 120448e8c6826b453d5b96e9076d61479a987677

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

@@ -5,10 +5,12 @@
import com.dy.common.mw.UnitStartedCallbackInterface;
import com.dy.common.threadPool.ThreadPool;
import com.dy.common.threadPool.TreadPoolFactory;
import com.dy.common.util.Callback;
import com.dy.testClient.ServerProperties;
import com.dy.testClient.rmiClient.RmiClUnit;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.mina.core.session.IoSession;

import java.util.Collection;
import java.util.HashMap;
@@ -26,7 +28,7 @@
    private static ThreadPool.Pool pool ;
    private static Map<String, MyThreadJob> jobMap = new HashMap<>() ;

    private static Integer totalRunedClientCount = 0;
    private static Integer totalRtuClientCount = 0;
    private static Integer totalSendDataCount = 0;
    private static Integer totalOverClientCount = 0;

@@ -72,23 +74,46 @@
                        if(!ServerProperties.startWork){
                            Thread.sleep(100L);
                        }else{
                            try{
                            startTime = System.currentTimeMillis() ;
                            for(Long addr = ServerProperties.rtuAddrStart; addr <= ServerProperties.rtuAddrEnd; addr++){
                                totalRunedClientCount++ ;
                                    totalRtuClientCount++ ;
                                createImitate(addr) ;
                            }
                                log.info("共模拟了" + totalRtuClientCount + "台RTU");

                                Collection<MyThreadJob> collection = jobMap.values() ;
                                for(MyThreadJob job : collection){
                                    connectServer(job) ;
                                }
                                log.info("启动所有RTU连接通信中间件");

                                while (true){
                                    int noConnectedCount = checkConnected() ;
                                    if(noConnectedCount > 0){
                                        log.info("等待" + noConnectedCount + "台RTU连接网络");
                                        Thread.sleep(100L);
                                    }else{
                                        break ;
                                    }
                                }

                            startJob() ;
                            while(true){
                                if(totalOverClientCount.longValue() >= totalRunedClientCount.longValue()){
                                    if(totalOverClientCount.longValue() >= totalRtuClientCount.longValue()){
                                    Long seconds = (System.currentTimeMillis() - startTime)/1000 ;
                                    RmiClUnit.getInstance().allOver(seconds) ;
                                    System.out.println("共用时" + seconds + "秒");
                                        log.info("共用时" + seconds + "秒");
                                    break ;
                                }else{
                                    Thread.sleep(100L);
                                }
                            }
                            }catch (Exception e){
                                e.printStackTrace();
                            }finally {
                            break;
                            }
                        }
                    }
                } catch (Exception e) {
@@ -106,18 +131,71 @@
        jobMap.put("" + rtuAddr, new MyThreadJob("" + rtuAddr, ServerProperties.tcpServerIp, ServerProperties.tcpServerPort)) ;
    }


    private void connectServer(MyThreadJob job){
        if(job.session == null){
            try{
                new TcpConnect().createSession(job.rtuAddr,
                        job,
                        job.serverIp,
                        job.serverPort,
                        job.connectTimeout,
                        new TcpHandler(),
                        new Callback() {
                            @Override
                            public void call(Object obj) {
                                if(obj == null){
                                    log.error("创建网络会话返回为null");
                                }else{
                                    job.session = (IoSession)obj ;
                                }
                            }
                            @Override
                            public void call(Object... objs) {
                            }
                            @Override
                            public void exception(Exception e) {
                            }
                        }) ;
            }catch (Exception e){
                job.exceptionOnConnect = true ;
                e.printStackTrace();
            }
        }
    }

    private int checkConnected(){
        int noConnectedCount = 0 ;
        Collection<MyThreadJob> collection = jobMap.values() ;
        for(MyThreadJob job : collection){
            if(job.session == null && !job.exceptionOnConnect){
                noConnectedCount++ ;
            }
        }
        return noConnectedCount;
    }

    private void startJob(){
        new Thread(new Runnable(){
            @Override
            public void run() {
                try {
                    Thread.sleep(1000L);
                    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 ;
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
@@ -129,17 +207,37 @@

    public static synchronized void clientSendData(){
        totalSendDataCount++;
        if(totalOverClientCount.longValue() >= totalRtuClientCount.longValue()){
            RmiClUnit.getInstance().reportHadReportCount(totalSendDataCount);
            System.out.println("已经发送" + totalSendDataCount + "条数据");
        }else{
            if(totalRtuClientCount > 100){
        if(totalSendDataCount % 100 == 0){
            RmiClUnit.getInstance().reportHadReportCount(totalSendDataCount);
            System.out.println("已经发送" + totalSendDataCount + "条数据");
                }
            }else{
                RmiClUnit.getInstance().reportHadReportCount(totalSendDataCount);
                System.out.println("已经发送" + totalSendDataCount + "条数据");
            }
        }
    }

    public static synchronized void clientOver(){
        totalOverClientCount++;
        if(totalOverClientCount.longValue() >= totalRtuClientCount.longValue()){
            RmiClUnit.getInstance().reportHadReportOver(totalOverClientCount);
            System.out.println("已有" + totalOverClientCount + "个RTU完成了任务");
        }else{
            if(totalRtuClientCount > 100) {
        if(totalOverClientCount % 100 == 0){
            RmiClUnit.getInstance().reportHadReportOver(totalOverClientCount);
            System.out.println("已有" + totalOverClientCount + "完成了任务");
                    System.out.println("已有" + totalOverClientCount + "个RTU完成了任务");
                }
            }else{
                RmiClUnit.getInstance().reportHadReportOver(totalOverClientCount);
                System.out.println("已有" + totalOverClientCount + "个RTU完成了任务");
            }
        }
    }
}