宽松模式多线程优化

朱兆平
src/main/java/com/sy/IMF/KAKO_Reader.java
src/main/java/com/sy/bwAnalysis/AnalysisRoute.java
src/main/java/com/sy/bwAnalysis/MessageAnalysis.java
src/main/java/com/sy/utils/XMLThreadFactory.java
src/main/java/com/sy/utils/XMLThreadPoolFactory.java
src/test/java/com/sy/AnalysisImfApplicationTests.java
src/test/java/com/sy/BeihuoTests.java
--- a/src/main/java/com/sy/IMF/KAKO_Reader.java
查看文件 @fe9283c
+++ b/src/main/java/com/sy/IMF/KAKO_Reader.java
查看文件 @fe9283c
@@ -2,9 +2,13 @@ package com.sy.IMF;
 import com.caac.imf.api.IMFClient;
 import com.sy.bwAnalysis.AnalysisRoute;
+import com.sy.utils.XMLThreadPoolFactory;
 import org.apache.commons.lang.StringUtils;
 import org.apache.log4j.Logger;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
 public class KAKO_Reader extends Thread{
     protected static final Logger logger = Logger.getLogger(KAKO_Reader.class);
     private IMFClient client;
@@ -22,13 +26,23 @@ public class KAKO_Reader extends Thread{
             logger.info("********读取线程状态true**********");
             logger.info("********Client-INFO= "+client+"********");
             Thread t =Thread.currentThread();
+            ThreadPoolExecutor threadPoolEs = XMLThreadPoolFactory.instance("kakou");
             while(true) {
                 if (IMF_Tesk.LOGIN_OK){
                     String message = this.client.getMSG();
                     logger.info(t.toString()+"读取线程已获取到消息");
                     if (message != null && StringUtils.isNotEmpty(message)) {
                      //   logger.info(message);
-                        AnalysisRoute.analysis(message);
+
+                        AnalysisRoute analysisRoute=new AnalysisRoute();
+                        analysisRoute.setMessage(message);
+                        threadPoolEs.execute(analysisRoute);
+                        logger.info("[THREADPOOL-INFO]-当前运行线程总数量: " + threadPoolEs.getActiveCount());
+                        logger.info("[THREADPOOL-INFO]-线程队列数量: " + threadPoolEs.getQueue().size());
+                        logger.info("[THREADPOOL-INFO]-完成的线程总数量: " + threadPoolEs.getCompletedTaskCount());
+                        logger.info("[THREADPOOL-INFO]-空闲线程释放时间(秒): " + threadPoolEs.getKeepAliveTime(TimeUnit.SECONDS));
+
+                        //AnalysisRoute.analysis(message);
 //                        if(message.indexOf("<TYPE>CARM</TYPE>") > 0){
 //                            this.client.sendMSG();
 //                        }
@@ -41,7 +55,7 @@ public class KAKO_Reader extends Thread{
                 }
                 try {
-                    Thread.sleep(500L);
+                    Thread.sleep(200L);
                 } catch (InterruptedException var3) {
                     logger.info("********读取线程循环获取消息异常---->"+var3.toString());
                     var3.printStackTrace();
--- a/src/main/java/com/sy/bwAnalysis/AnalysisRoute.java
查看文件 @fe9283c
+++ b/src/main/java/com/sy/bwAnalysis/AnalysisRoute.java
查看文件 @fe9283c
@@ -15,9 +15,8 @@ import javax.annotation.PostConstruct;
 import java.util.Date;
 import java.util.List;
-
 @Component
-public class AnalysisRoute {
+public class AnalysisRoute implements Runnable{
     @Autowired
     private aironeExStockService exStockService;
@@ -50,14 +49,15 @@ public class AnalysisRoute {
     private static AnalysisRoute route;
-    private static MessageAnalysis analysis = new MessageAnalysis();
-    private static CLRAnalysis clrAnalysis = new CLRAnalysis();
-    private static ExStockAnalysis exStockAnalysis = new ExStockAnalysis();
-    private static ImStockAnalysis imStockAnalysis = new ImStockAnalysis();
-    private static GATAnaluysis gatAnaluysis = new GATAnaluysis();
-    private static GatherInfoAnalysis gatherInfoAnalysis = new GatherInfoAnalysis();
-    private static CommandInfoAnalysis commandInfoAnalysis = new CommandInfoAnalysis();
-    private static ResMessageAnalysis resMessageAnalysis = new ResMessageAnalysis();
+    public String getMessage() {
+        return message;
+    }
+
+    public void setMessage(String message) {
+        this.message = message;
+    }
+
+    private String message;
     @PostConstruct
     public void init(){
@@ -77,7 +77,16 @@ public class AnalysisRoute {
      * 解析从IMF接收过来的报文
      * @param string 报文内容
      */
-    public static void analysis(String string) {
+    public void analysis(String string) {
+        MessageAnalysis analysis = new MessageAnalysis();
+        CLRAnalysis clrAnalysis = new CLRAnalysis();
+        ExStockAnalysis exStockAnalysis = new ExStockAnalysis();
+        ImStockAnalysis imStockAnalysis = new ImStockAnalysis();
+        GATAnaluysis gatAnaluysis = new GATAnaluysis();
+        GatherInfoAnalysis gatherInfoAnalysis = new GatherInfoAnalysis();
+        CommandInfoAnalysis commandInfoAnalysis = new CommandInfoAnalysis();
+        ResMessageAnalysis resMessageAnalysis = new ResMessageAnalysis();
+
         PropertyConfigurator.configure("config/log4j.properties");
         string = string.replace("Msg","MSG");
         Message message = analysis.readTicketsXml(string);;
@@ -199,6 +208,10 @@ public class AnalysisRoute {
     }
+    @Override
+    public void run() {
+        analysis(message);
+    }
 }
--- a/src/main/java/com/sy/bwAnalysis/MessageAnalysis.java
查看文件 @fe9283c
+++ b/src/main/java/com/sy/bwAnalysis/MessageAnalysis.java
查看文件 @fe9283c
@@ -7,7 +7,6 @@ import com.thoughtworks.xstream.io.xml.XmlFriendlyNameCoder;
 public class MessageAnalysis {
-    public static Message message = null;
     public Message readTicketsXml(String str) {
         return getMessage(str);
@@ -20,8 +19,7 @@ public class MessageAnalysis {
         XStream.setupDefaultSecurity(xstream);
         //对xstream对象设置默认的安全防护时,允许设置类
         xstream.allowTypes(new Class[]{Message.class});
-        message = (Message) xstream.fromXML(str);
-        return message;
+        return (Message) xstream.fromXML(str);
     }
--- a/src/main/java/com/sy/utils/XMLThreadFactory.java 0 → 100644
查看文件 @fe9283c
+++ b/src/main/java/com/sy/utils/XMLThreadFactory.java 0 → 100644
查看文件 @fe9283c
+package com.sy.utils;
+
+
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.ThreadFactory;
+
+public class XMLThreadFactory implements ThreadFactory {
+
+    private int          counter;
+    private String       name;
+    private List<String> stats;
+
+    public XMLThreadFactory(String name)
+    {
+        counter = 1;
+        this.name = name;
+        stats = new ArrayList<String>();
+    }
+
+    @Override
+    public Thread newThread(Runnable runnable)
+    {
+        Thread t = new Thread(runnable, name + "-Thread_" + counter);
+        counter++;
+        stats.add(String.format("Created thread %d with name %s on %s \n", t.getId(), t.getName(), new Date()));
+        //log.info("Created thread id: {} with name {} on {} \n", t.getId(), t.getName(), new Date());
+        //log.info(getStats());
+        return t;
+    }
+
+    public String getStats()
+    {
+        StringBuffer buffer = new StringBuffer();
+        Iterator<String> it = stats.iterator();
+        while (it.hasNext())
+        {
+            buffer.append(it.next());
+        }
+        return buffer.toString();
+    }
+
+}
--- a/src/main/java/com/sy/utils/XMLThreadPoolFactory.java 0 → 100644
查看文件 @fe9283c
+++ b/src/main/java/com/sy/utils/XMLThreadPoolFactory.java 0 → 100644
查看文件 @fe9283c
+package com.sy.utils;
+
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+public class XMLThreadPoolFactory {
+
+    private  static ThreadPoolExecutor threadPool;
+
+    public static ThreadPoolExecutor instance(String busstype){
+        if (threadPool==null){
+            XMLThreadFactory xmlThreadFactory = new XMLThreadFactory(busstype);
+            threadPool = new ThreadPoolExecutor(50, 200,
+                    1L, TimeUnit.SECONDS,
+                    new LinkedBlockingQueue<Runnable>(4000),
+                    xmlThreadFactory,
+                    new ThreadPoolExecutor.AbortPolicy());
+
+        }
+        return threadPool;
+    }
+}
--- a/src/test/java/com/sy/AnalysisImfApplicationTests.java
查看文件 @fe9283c
+++ b/src/test/java/com/sy/AnalysisImfApplicationTests.java
查看文件 @fe9283c
@@ -2,21 +2,26 @@ package com.sy;
 import com.sy.bwAnalysis.AnalysisRoute;
 import com.sy.logic.LiftBar;
+import com.sy.utils.XMLThreadPoolFactory;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.test.context.junit4.SpringRunner;
+import java.util.concurrent.ThreadPoolExecutor;
+
 @RunWith(SpringRunner.class)
 @SpringBootTest
 public class AnalysisImfApplicationTests {
     private static String message = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n" +
             "<MSG><META><SNDR>KAO</SNDR><RCVR></RCVR><SEQN>20220117192051</SEQN><DDTM>20220117192051</DDTM><TYPE>KAKO</TYPE><STYP>CARM</STYP></META><COMMAND_INFO AREA_ID=\"4604499001\" CHNL_NO=\"4604444412\" I_E_TYPE=\"E\" SEQ_NO=\"20220117192014000017\"><CHECK_RESULT>00000000100000000000</CHECK_RESULT><OP_HINT>直接放行</OP_HINT><SEAL><ESEAL_ID>CNSM2816998419</ESEAL_ID><SEAL_KEY>1234567890</SEAL_KEY><OPEN_TIMES/><ESEAL_IC_NO/></SEAL><SZ_MSG/></COMMAND_INFO></MSG>";
-
+    private static ThreadPoolExecutor threadPoolEs = XMLThreadPoolFactory.instance("kakou");
+    private AnalysisRoute analysisRoute=new AnalysisRoute();
     @Test
     public void contextLoads() {
-        AnalysisRoute.analysis(message);
+        analysisRoute.setMessage(message);
+        threadPoolEs.execute(analysisRoute);
         LiftBar.sendData("1","aaa",true);
--- a/src/test/java/com/sy/BeihuoTests.java
查看文件 @fe9283c
+++ b/src/test/java/com/sy/BeihuoTests.java
查看文件 @fe9283c
@@ -2,18 +2,26 @@ package com.sy;
 import com.sy.bwAnalysis.AnalysisRoute;
 import com.sy.logic.LiftBar;
+import com.sy.utils.XMLThreadPoolFactory;
 import org.apache.log4j.Logger;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.test.context.junit4.SpringRunner;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
 @RunWith(SpringRunner.class)
 @SpringBootTest
 public class BeihuoTests {
+
     private static final Logger logger = Logger.getLogger(LiftBar.class);
+    private static ThreadPoolExecutor threadPoolEs = XMLThreadPoolFactory.instance("kakou");
+    private AnalysisRoute analysisRoute=new AnalysisRoute();
+
     //车牌号
     private String voNo = "豫A61CR7";
     //备案重量
@@ -98,7 +106,9 @@ public class BeihuoTests {
                 .replace("${IETYPE}",ie)
                 .replace("${barcode}",barcode);
         //进港提货测试
-        AnalysisRoute.analysis(IMPORT_XML);
+
+        analysisRoute.setMessage(IMPORT_XML);
+
     }
     /**
@@ -129,7 +139,14 @@ public class BeihuoTests {
                 .replace("${IETYPE}",ie)
                 .replace("${barcode}",barcode);
         //进港提货测试
-        AnalysisRoute.analysis(IMPORT_XML);
+        analysisRoute.setMessage(IMPORT_XML);
+        threadPoolEs.execute(analysisRoute);
+
+        System.out.println("[THREAD-INFI]-线程运行线程总数量 = " + threadPoolEs.getActiveCount());
+        System.out.println("[THREAD-INFI]-线程队列数量 = " + threadPoolEs.getQueue().size());
+        System.out.println("threadPoolEs.getCompletedTaskCount() = " + threadPoolEs.getCompletedTaskCount());
+        System.out.println("threadPoolEs.getKeepAliveTime(TimeUnit.SECONDS) = " + threadPoolEs.getKeepAliveTime(TimeUnit.SECONDS));
+        System.out.println("threadPoolEs.getTaskCount() = " + threadPoolEs.getTaskCount());
     }