update...

潘钊
1 parent 3f23f994
Showing 4 changed files with 49 additions and 37 deletions
src/main/java/com/bsth/client/GpsBeforeBuffer.java
src/main/java/com/bsth/client/pd/handler/PdClientHandler.java
src/main/java/com/bsth/client/pf/handler/PfClientHandler.java
src/main/java/com/bsth/entity/GpsEntity.java
@@ -10,12 +10,8 @@ import org.springframework.core.annotation.Order;
 import org.springframework.stereotype.Component;
 import java.util.ArrayList;
-import java.util.LinkedList;
 import java.util.List;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 /**
  * 从 socket client 到调度系统 的缓冲
@@ -25,14 +21,23 @@ import java.util.concurrent.TimeUnit;
 @Order(2)
 public class GpsBeforeBuffer implements CommandLineRunner{
-    static LinkedList<GpsEntity> linkedList = new LinkedList();
+    static ConcurrentLinkedQueue<GpsEntity> linkedList = new ConcurrentLinkedQueue();
     static final int MAX_SIZE = 4000 * 20;
+    static int size = 0;
     public void put(GpsEntity gps){
-        if(gps == null || StringUtils.isBlank(gps.getDeviceId()))
-            return;
+        try {
+            if(gps == null)
+                return;
+
+            if(StringUtils.isBlank(gps.getDeviceId()))
+                return;
-        linkedList.add(gps);
+            linkedList.add(gps);
+            size++;
+        }catch (Exception e){
+            log.error("", e);
+        }
     }
     static Logger log = LoggerFactory.getLogger(GpsBeforeBuffer.class);
@@ -40,23 +45,17 @@ public class GpsBeforeBuffer implements CommandLineRunner{
     public static List<GpsEntity> pollAll(){
         List<GpsEntity> rs = new ArrayList<>(300);
         GpsEntity gps;
-        int size = linkedList.size();
-        for(int j = 0; j < size; j++){
-            gps = linkedList.poll();
-            if(gps == null){
-                log.error("linkedList poll gps is null ???");
-            }
-            rs.add(gps);
-        }
-        /*while (true){
+
+        while (true){
             gps = linkedList.poll();
             if(gps == null){
+                size = 0;
                 break;
             }
             rs.add(gps);
-        }*/
+        }
-        log.info("poll size: " + rs.size());
+        log.info("poll size: " + rs.size() + " -current size: " + size);
         return rs;
     }
@@ -64,33 +63,27 @@ public class GpsBeforeBuffer implements CommandLineRunner{
      * 清理数据，保持最大 MAX_SIZE 个数的元素
      */
     public static void clear(){
-        int size = linkedList.size();
         if(size <= MAX_SIZE)
             return;
-
         int len = size - MAX_SIZE;
         for(int j = 0; j < len; j++){
             linkedList.poll();
+            size--;
         }
-        log.info("clear size: " + len);
+        log.info("clear size: " + len + " -current size: " + size);
     }
     @Autowired
     BufferSizeCheck bufferSizeCheck;
+    @Autowired
+    SizeCheck sizeCheck;
-    ScheduledExecutorService sexec;
+    ScheduledExecutorService sexec = Executors.newScheduledThreadPool(2);
     @Override
     public void run(String... strings) throws Exception {
-        sexec = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
-            @Override
-            public Thread newThread(Runnable r) {
-                Thread t = new Thread(r);
-                t.setName("gpsBufferSizeCheck");
-                return t;
-            }
-        });
         sexec.scheduleWithFixedDelay(bufferSizeCheck, 60, 30, TimeUnit.SECONDS);
+        sexec.scheduleWithFixedDelay(sizeCheck, 60, 60 * 5, TimeUnit.SECONDS);
     }
     @Component
@@ -101,4 +94,13 @@ public class GpsBeforeBuffer implements CommandLineRunner{
             GpsBeforeBuffer.clear();
         }
     }
+
+    @Component
+    public static class SizeCheck extends Thread{
+
+        @Override
+        public void run() {
+            log.info("[SizeCheck] linkedList real size: " + linkedList.size() + " -current size: " + size);
+        }
+    }
 }
 \ No newline at end of file
@@ -73,11 +73,17 @@ public class PdClientHandler extends IoHandlerAdapter{
         	}
             else if(0x41 == msg.getCommandType()){
                 Pd_41_0 pd41 = (Pd_41_0)msg.getMessageBody();
-                gpsBeforeBuffer.put(GpsEntity.getInstance(pd41.getInfo(), msg.getVersion(), 1));
+
+                GpsEntity gps = GpsEntity.getInstance(pd41.getInfo(), msg.getVersion(), 1);
+                if(gps != null)
+                    gpsBeforeBuffer.put(gps);
             }
             else if(0x42 == msg.getCommandType()){
                 Pd_42_0 pd42 = (Pd_42_0)msg.getMessageBody();
-                gpsBeforeBuffer.put(GpsEntity.getInstance(pd42.getInfo(), msg.getVersion(), 1));
+
+                GpsEntity gps = GpsEntity.getInstance(pd42.getInfo(), msg.getVersion(), 1);
+                if(gps != null)
+                    gpsBeforeBuffer.put(GpsEntity.getInstance(pd42.getInfo(), msg.getVersion(), 1));
             }
         }
     }
@@ -65,13 +65,16 @@ public class PfClientHandler extends IoHandlerAdapter{
         PfMessage msg = (PfMessage)message;
         IMessageBody body = msg.getMessageBody();
         if (body != null) {
-        	String deviceId = body.getDeviceId();
+        	//String deviceId = body.getDeviceId();
         	if (0x1 == msg.getCommandType()) {
         		log.debug("设备编号:" + body.getDeviceId() + "建立连接");
         	}
 			BasicInfo info = Protocol2BizUtil.getBasicInfoFromMsg(msg);
-			gpsBeforeBuffer.put(GpsEntity.getInstance(info, msg.getVersion(), 0));
+
+			GpsEntity gps = GpsEntity.getInstance(info, msg.getVersion(), 0);
+			if(gps != null)
+				gpsBeforeBuffer.put(gps);
         }
     }
 package com.bsth.entity;
 import com.bsth.client.pd.protocol.BasicInfo;
+import org.apache.commons.lang3.StringUtils;
 /**
  * @author PanZhao
@@ -77,7 +78,7 @@ public class GpsEntity {
     public static GpsEntity getInstance(BasicInfo basicInfo, int version, int source){
         //放弃补发数据
         byte cacheData = getCacheState(basicInfo.getServiceState());
-        if(cacheData == 1)
+        if(cacheData == 1 || StringUtils.isBlank(basicInfo.getDeviceId()))
             return null;
         GpsEntity gps = new GpsEntity();