线程池修改

src/main/java/com/chinaitop/depot/intelligent/grainsituation/mapper/TGasDataMapper.xml

@@ -413,9 +413,11 @@
                 and store_code = #{storeCode}
               </if>
             </where>
+            GROUP BY store_code
         ) tg
     INNER JOIN t_gas_data tgd ON tg.result_time = tgd.result_time
     AND tg.org_id = tgd.org_id
     AND tg.store_code = tgd.store_code
+    ORDER BY tgd.store_code
   </select>
 </mapper>

src/main/java/com/chinaitop/depot/intelligent/grainsituation/service/impl/GasDetectionServiceImpl.java

@@ -1,9 +1,6 @@
 package com.chinaitop.depot.intelligent.grainsituation.service.impl;
 
-import com.chinaitop.depot.intelligent.basicdata.mapper.TDcsqtinfoMapper;
 import com.chinaitop.depot.intelligent.basicdata.mapper.TQtdevinfoMapper;
-import com.chinaitop.depot.intelligent.basicdata.model.TDcsqtinfo;
-import com.chinaitop.depot.intelligent.basicdata.model.TDcsqtinfoExample;
 import com.chinaitop.depot.intelligent.basicdata.model.TQtdevinfo;
 import com.chinaitop.depot.intelligent.basicdata.model.TQtdevinfoExample;
 import com.chinaitop.depot.intelligent.common.ConstantClass;
@@ -20,15 +17,13 @@ import com.chinaitop.depot.intelligent.utils.UuidUtils;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import org.apache.commons.lang3.StringUtils;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
-import reactor.util.annotation.Nullable;
 
 import javax.annotation.Resource;
 import java.util.*;
 
-@Service
+@Service("gasDetectionService")
 public class GasDetectionServiceImpl implements GasDetectionService {
     @Resource
     private TQtdevinfoValueMapper tQtdevinfoValueMapper;
@@ -121,6 +116,7 @@ public class GasDetectionServiceImpl implements GasDetectionService {
             Date endDate = DateUtils.stringToData(stringFactory.joinString(1, endTime, DateUtils.endTime));
             criteria.andResultTimeBetween(startDate, endDate);
         }
+        example.setOrderByClause("result_time desc");
         return gasDataMapper.selectByExample(example);
     }
 
@@ -141,29 +137,28 @@ public class GasDetectionServiceImpl implements GasDetectionService {
             String storeCode = gasConfig.getStoreCode();
             // 数据list
             ArrayList<TGasData> gases = Lists.newArrayList();
-            objects1.forEach(data -> {
-                if (data.size() > 12) {
-                    TGasData gasData = new TGasData();
-                    gasData.setNowO2(data.get(0)); // 氧气实时
-                    gasData.setNowPh3(data.get(1)); // 磷化氢实时
-                    gasData.setSouthO2(data.get(2)); // 南检测结果 氧气
-                    gasData.setSouthPh3(data.get(3)); // 南检测结果 磷化氢
-                    gasData.setEastO2(data.get(4)); // 东检测结果 氧气
-                    gasData.setEastPh3(data.get(5)); // 东检测结果 磷化氢
-                    gasData.setNorthO2(data.get(6)); // 北检测结果 氧气
-                    gasData.setNorthPh3(data.get(7)); // 北检测结果 磷化氢
-                    gasData.setWestO2(data.get(8)); // 西检测结果 氧气
-                    gasData.setWestPh3(data.get(9)); // 西检测结果 磷化氢
-                    gasData.setInO2(data.get(10)); // 中检测结果 氧气
-                    gasData.setInPh3(data.get(11)); // 中检测结果 磷化氢
-                    gases.add(gasData);
-                }
-            });
+            for (List<Float> data : objects1) {
+                TGasData gasData = new TGasData();
+                gasData.setNowPh3(data.get(0)); // 磷化氢实时
+                gasData.setNowO2(data.get(1)); // 氧气实时
+                gasData.setSouthO2(data.get(2)); // 南检测结果 氧气
+                gasData.setSouthPh3(data.get(3)); // 南检测结果 磷化氢
+                gasData.setEastO2(data.get(4)); // 东检测结果 氧气
+                gasData.setEastPh3(data.get(5)); // 东检测结果 磷化氢
+                gasData.setNorthO2(data.get(6)); // 北检测结果 氧气
+                gasData.setNorthPh3(data.get(7)); // 北检测结果 磷化氢
+                gasData.setWestO2(data.get(8)); // 西检测结果 氧气
+                gasData.setWestPh3(data.get(9)); // 西检测结果 磷化氢
+                gasData.setInO2(data.get(10)); // 中检测结果 氧气
+                gasData.setInPh3(data.get(11)); // 中检测结果 磷化氢
+                gases.add(gasData);
+            }
             // 保存对象
             TGasData saveOnlyGasData = new TGasData();
             saveOnlyGasData.setId(UuidUtils.getCode()); // id
             saveOnlyGasData.setOrgId(orgId); // 组织机构编码
             saveOnlyGasData.setStoreCode(storeCode); // 仓房编码
+            saveOnlyGasData.setResultTime(DateUtils.getDate());
             // 计算平均值
             float avgNowO2 = (float) gases.stream().mapToDouble(TGasData::getNowO2).average().getAsDouble();
             float avgNowPh3 = (float) gases.stream().mapToDouble(TGasData::getNowPh3).average().getAsDouble();

src/main/java/com/chinaitop/depot/intelligent/socket/ServerThread.java

@@ -2,6 +2,7 @@ package com.chinaitop.depot.intelligent.socket;
 
 import com.alibaba.fastjson.JSON;
 import com.chinaitop.depot.intelligent.grainsituation.service.GasDetectionService;
+import com.chinaitop.depot.intelligent.utils.GetBeanUtil;
 import com.google.common.collect.Lists;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -21,8 +22,6 @@ public class ServerThread implements Runnable {
     private final static Logger log = LoggerFactory.getLogger(ServerThread.class);
 
     private Socket socket;
-    @Resource
-    private GasDetectionService gasDetectionService;
 
     ServerThread(Socket socket) {
         this.socket = socket;
@@ -30,17 +29,17 @@ public class ServerThread implements Runnable {
 
     @Override
     public void run() {
-        InputStream inputStream = null; // 字节输入流
+//        InputStream inputStream = null; // 字节输入流
         try {
             synchronized (this) {
-                inputStream = socket.getInputStream();
+                InputStream inputStream = socket.getInputStream();
                 int read; // 读取数据变量
                 // 缓存返回数据
                 Integer[] data = null; // 动态扩容数组初始化
-                // 获取5983次实时数据,初始化
+                // 获取5967次实时数据,初始化
                 int count = 0;
-                // 获取数据次数位 117位*51+16 为一组数据
-                int whileCount = 0X175F;
+                // 获取数据次数位 117位*51 为一组数据
+                int whileCount = 0X174F;
                 //循环读取客户端信息
                 while (true) {
                     count++;
@@ -48,7 +47,7 @@ public class ServerThread implements Runnable {
                     read = inputStream.read();
                     //获取客户端的ip地址及发送数据
 //                    System.out.println("服务器端接收：" + "{'from_client':'" + socket.getInetAddress().getHostAddress() + "','data':'" + read + "'}");
-//                    System.out.println("服务器端接收：" + read);
+//                    System.out.println(read);
 
                     // 动态扩容数组
                     if (data == null) { //若数组为空，定义数组的长度为1
@@ -94,9 +93,9 @@ public class ServerThread implements Runnable {
         } finally {
             //关闭资源
             try {
-                if (inputStream != null) {
+                /*if (inputStream != null) {
                     inputStream.close();
-                }
+                }*/
                 if (socket != null) {
                     socket.close();
                 }
@@ -107,16 +106,14 @@ public class ServerThread implements Runnable {
     }
 
     private void gasDataHandle(Integer[] data, String ip) {
-        // 1.将数据前16位排除
-        Integer[] integers = Arrays.copyOfRange(data, 16, data.length);
         // 2.将数据分成51组,每组117个位
         int bitFromData = 0;
         int bitToData = 117;
         Integer[] groupData;
         List<Integer[]> list = new ArrayList<>();
-        for (int i = 0; i < integers.length; i++) {
-            if (i % 116 == 0 && bitToData <= integers.length) {
-                groupData = Arrays.copyOfRange(integers, bitFromData, bitToData);
+        for (int i = 0; i < data.length; i++) {
+            if (i % 116 == 0 && bitToData <= data.length) {
+                groupData = Arrays.copyOfRange(data, bitFromData, bitToData);
                 bitFromData += 117;
                 bitToData += 117;
                 list.add(groupData);
@@ -163,6 +160,7 @@ public class ServerThread implements Runnable {
         });
         log.info("处理后的PLC测气返回数据: {}", JSON.toJSON(objects1));
         // 将数据保存到数据库
+        GasDetectionService gasDetectionService = (GasDetectionService) GetBeanUtil.getBean("gasDetectionService");
         gasDetectionService.saveGasData(objects1, ip);
     }
 

src/main/java/com/chinaitop/depot/intelligent/socket/SocketServer.java

@@ -40,6 +40,8 @@ public class SocketServer {
     private static final int tempCount = 0x40;
     // 本类注入ServerSocket
     private ServerSocket serverSocket;
+    private ExecutorService executorService;//线程池
+    private final int PoolSize = 8;//单个CPU线程池大小
 
     public SocketServer() throws IOException {
         /*
@@ -54,6 +56,9 @@ public class SocketServer {
          */
         InetAddress localHost = InetAddress.getLocalHost(); // 获取本机ip(监听地址为服务器地址需要PLC修改)
         serverSocket = new ServerSocket(port, backlog, localHost); // 连接请求队列的长度为3
+        // //Runtime的availableProcessor()方法返回当前系统的CPU数目
+        int i = Runtime.getRuntime().availableProcessors();
+        executorService = Executors.newFixedThreadPool(i * PoolSize);
         log.info("SocketServer服务器启动!");
     }
 
@@ -64,7 +69,7 @@ public class SocketServer {
      *             TODO 仓上控制柜 1:测气,0:自然窗,0:轴流窗,0:轴流风机,
      *             TODO 仓下控制柜 0:通风口
      */
-    public String service(String ip, String orders, int type) {
+    public /*synchronized*/ String service(String ip, String orders, int type) {
         String resultPLC = "";
         Socket socket;
         try {
@@ -92,10 +97,15 @@ public class SocketServer {
                     }
                     if (type == 1 && !orders.equals(ConstantClass.TYPE_ON)) {
                         // 创建线程池对象 // TODO 优化方案 1.多线程使用线程池,防止服务器宕机
-                        ExecutorService executorService = Executors.newFixedThreadPool(8); // 创建固定大小线程池
+
+                        /*ExecutorService executorService = Executors.newFixedThreadPool(16); // 创建固定大小线程池
                         ServerThread serverThread = new ServerThread(socket);
                         // 启动线程等待返回(测气)
-                        executorService.submit(serverThread);
+                        executorService.submit(serverThread);*/
+
+//                        executorService.execute(new ServerThread(socket));
+                        executorService.submit(new ServerThread(socket));
+
 //                        new Thread(new ServerThread(socket)).start(); // 启动线程等待返回(测气)
                         resultPLC = ConstantClass.C_SOH_S;
                     } else {

src/main/java/com/chinaitop/depot/intelligent/utils/GetBeanUtil.java

@@ -0,0 +1,27 @@
+package com.chinaitop.depot.intelligent.utils;
+
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author qingsong.han
+ * @description:
+ * @create 2021-10-17 12:40
+ */
+@Component
+public class GetBeanUtil implements ApplicationContextAware {
+    private static ApplicationContext applicationContext;
+
+    public void setApplicationContext(ApplicationContext context) {
+        GetBeanUtil.applicationContext = context;
+    }
+
+    public static Object getBean(String name) {
+        return applicationContext.getBean(name);
+    }
+
+    public static ApplicationContext getApplicationContext() {
+        return applicationContext;
+    }
+}