update...

潘钊
1 parent 460b9760
Showing 11 changed files with 120 additions and 39 deletions
src/main/java/com/bsth/controller/realcontrol/anomalyCheckController.java
src/main/java/com/bsth/data/car_out_info/CarOutInfoHandler.java
src/main/java/com/bsth/data/gpsdata/arrival/GpsRealAnalyse.java
src/main/java/com/bsth/data/gpsdata/client/ClientApp.java
src/main/java/com/bsth/data/gpsdata/client/DataMonitor.java
src/main/java/com/bsth/data/gpsdata/client/GpsBeforeBuffer.java
src/main/java/com/bsth/data/gpsdata/client/pd/codec/MessageDecoder.java
src/main/java/com/bsth/data/gpsdata/client/pd/common/ConvertUtil.java
src/main/java/com/bsth/data/gpsdata/client/pd/handler/PdClientHandler.java
src/main/java/com/bsth/data/gpsdata/client/pf/codec/MessageDecoder.java
src/main/java/com/bsth/data/gpsdata/client/pf/handler/PfClientHandler.java
@@ -56,4 +56,14 @@ public class anomalyCheckController {
         ClientApp.pdreconn();
         ClientApp.pfreconn();
     }
+
+    @RequestMapping(value = "/pdClose", method = RequestMethod.POST)
+    public void pdClose(){
+        ClientApp.pdClose();
+    }
+
+    @RequestMapping(value = "/pfClose", method = RequestMethod.POST)
+    public void pfClose(){
+        ClientApp.pfClose();
+    }
 }
@@ -102,7 +102,7 @@ public class CarOutInfoHandler implements CommandLineRunner, CarOutInfo {
                 ps.setString(2, sch.getXlBm());
                 ps.setString(3, sch.getXlName());
                 ps.setString(4, sch.getLpName());
-                ps.setInt(5, sch.getFcno());
+                ps.setInt(5, sch.getFcno()==null?-1:sch.getFcno());
                 ps.setString(6, sch.getDfsj());
                 ps.setString(7, sch.getClZbh());
                 ps.setString(8, BasicData.nbbmCompanyPlateMap.get(sch.getClZbh()));
@@ -49,6 +49,8 @@ public class GpsRealAnalyse {
         //如果正在恢复数据
         if (GpsDataRecovery.run)
             return;
+
+        long t = System.currentTimeMillis();
         logger.info("analyse gps size: " + list.size());
         //按车辆分组gps
         ArrayListMultimap multimap = ArrayListMultimap.create();
@@ -70,6 +72,8 @@ public class GpsRealAnalyse {
             //加入实时gps对照
             for(GpsEntity gps: list)
                 gpsRealData.put(gps);
+
+            logger.info("time , " + (System.currentTimeMillis() - t));
         } catch (InterruptedException e) {
             logger.error("", e);
         }
@@ -21,10 +21,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 import java.net.InetSocketAddress;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 /**
  * Created by panzhao on 2017/5/4.
@@ -43,6 +40,7 @@ public class ClientApp {
     GpsBeforeBuffer gpsBeforeBuffer;
     static Logger logger = LoggerFactory.getLogger(ClientApp.class);
+    private static ExecutorService exec;
     private ScheduledExecutorService sexec = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
@@ -85,12 +83,52 @@ public class ClientApp {
         return flag;
     }
+    public static void pdconnect(final String device) {
+        exec.submit(new Runnable() {
+
+            @Override
+            public void run() {
+                // TODO Auto-generated method stub
+                long now = System.currentTimeMillis();
+                boolean flag = false;
+                while (!flag) {
+                    flag = dconnect(device);
+                }
+                System.out.println("设备编号:" + device + "重连, cost time: " + (System.currentTimeMillis() - now));
+            }
+        });
+    }
+
+    public static void pfconnect(final String device) {
+        exec.submit(new Runnable() {
+
+            @Override
+            public void run() {
+                // TODO Auto-generated method stub
+                long now = System.currentTimeMillis();
+                boolean flag = false;
+                while (!flag) {
+                    flag = fconnect(device);
+                }
+                System.out.println("重连, cost time: " + (System.currentTimeMillis() - now));
+            }
+        });
+    }
+
     public static void pdreconn(){
-        dconnect(ConfigUtil.get("forward.device.name"));
+        pdconnect(ConfigUtil.get("forward.device.name"));
     }
     public static void pfreconn(){
-        fconnect(ConfigUtil.get("forward.device.name"));
+        pfconnect(ConfigUtil.get("forward.device.name"));
+    }
+
+    public static void pdClose(){
+        pdSession.closeNow();
+    }
+
+    public static void pfClose(){
+        pfSession.closeNow();
     }
     public static boolean fconnect(String device) {
@@ -111,7 +149,7 @@ public class ClientApp {
     }
     public void init() {
-        //exec = Executors.newFixedThreadPool(4);
+        exec = Executors.newFixedThreadPool(50);
         sexec.scheduleAtFixedRate(new SessionChecker(), 1, 1, TimeUnit.MINUTES);
         /*******************************浦东********************************/
         pdDataConnector = new NioSocketConnector();
@@ -133,7 +171,7 @@ public class ClientApp {
         pdDataConnector.setHandler(pdClient);
-        dconnect(ConfigUtil.get("forward.device.name"));
+        pdconnect(ConfigUtil.get("forward.device.name"));
         /*******************************浦东转发********************************/
         pfDataConnector = new NioSocketConnector();
@@ -153,7 +191,7 @@ public class ClientApp {
         config1.setIdleTime(IdleStatus.BOTH_IDLE, 60);
         pfDataConnector.setHandler(pfClient);
-        fconnect(ConfigUtil.get("forward.device.name"));
+        pfconnect(ConfigUtil.get("forward.device.name"));
         gpsBeforeBuffer.init();
@@ -182,6 +220,5 @@ public class ClientApp {
                 logger.error("SessionChecker异常", e);
             }
         }
-
     }
 }
 \ No newline at end of file
-package com.bsth.data.gpsdata.client;
-
-/**
- * Created by panzhao on 2017/5/7.
- */
-public class DataMonitor {
-
-    public static long lastTimePd;
-
-    public static long lastTimePf;
-
-}
@@ -63,7 +63,7 @@ public class GpsBeforeBuffer {
     }
     public void init(){
-        Application.mainServices.scheduleWithFixedDelay(gpsHandleThread, 20 * 1000, 1100, TimeUnit.MILLISECONDS);
+        Application.mainServices.scheduleWithFixedDelay(gpsHandleThread, 20 * 1000, 1200, TimeUnit.MILLISECONDS);
     }
     @Component
@@ -79,7 +79,7 @@ public class GpsBeforeBuffer {
             list = new ArrayList<>(100);
             GpsEntity gps;
-            for(int i = 0; i < 2000; i ++){
+            for(int i = 0; i < 4000; i ++){
                 gps = linkedList.poll();
                 if(gps == null)
                     break;
 package com.bsth.data.gpsdata.client.pd.codec;
+import com.bsth.data.gpsdata.client.pd.common.ConvertUtil;
 import com.bsth.data.gpsdata.client.pd.protocol.PdMessage;
 import org.apache.mina.core.buffer.IoBuffer;
 import org.apache.mina.core.session.IoSession;
 import org.apache.mina.filter.codec.CumulativeProtocolDecoder;
 import org.apache.mina.filter.codec.ProtocolDecoderOutput;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 public class MessageDecoder extends CumulativeProtocolDecoder {
+	Logger log = LoggerFactory.getLogger(MessageDecoder.class);
+
 	@Override
 	protected boolean doDecode(IoSession session, IoBuffer in,
 			ProtocolDecoderOutput out) throws Exception {
@@ -19,15 +24,18 @@ public class MessageDecoder extends CumulativeProtocolDecoder {
 			int len = ((lenh & 0xff) << 8) + (lenl & 0xff);
 			if ((head1 & 0xff) == 0xfa && (head2 & 0xff) == 0xf5) {
 				if (in.remaining() > len) {
-					byte[] bytes = new byte[len + 1];
-					in.get(bytes);
-					PdMessage msg = new PdMessage();
-					msg.read(bytes);
-					out.write(msg);
-					/*if ("true".equals(ConfigUtil.getProperty("protocolup", "true"))) {
-						UpProtocolDataService.getInstance().write(new byte[]{ head1, head2, lenh, lenl });
-						UpProtocolDataService.getInstance().write(bytes);
-					}*/
+					try {
+						byte[] bytes = new byte[len + 1];
+						in.get(bytes);
+						PdMessage msg = new PdMessage();
+						msg.read(bytes);
+						out.write(msg);
+
+						//日志纪录
+						log.info("pd client receive: " + ConvertUtil.bytesToHexString(bytes));
+					}catch (Exception e){
+						log.error("pd message decoder：", e);
+					}
 				} else {
 					in.reset();
 					return false;
@@ -126,4 +126,24 @@ public class ConvertUtil {
         return val;
 	}
+
+	/* Convert byte[] to hex string.这里我们可以将byte转换成int，然后利用Integer.toHexString(int)来转换成16进制字符串。
+        * @param src byte[] data
+* @return hex string
+*/
+	public static String bytesToHexString(byte[] src){
+		StringBuilder stringBuilder = new StringBuilder("");
+		if (src == null || src.length <= 0) {
+			return null;
+		}
+		for (int i = 0; i < src.length; i++) {
+			int v = src[i] & 0xFF;
+			String hv = Integer.toHexString(v);
+			if (hv.length() < 2) {
+				stringBuilder.append(0);
+			}
+			stringBuilder.append(hv);
+		}
+		return stringBuilder.toString();
+	}
 }
@@ -52,6 +52,7 @@ public class PdClientHandler extends IoHandlerAdapter{
 	@Override
     public void sessionIdle(IoSession session, IdleStatus status) throws Exception {
+        log.warn("pd sessionIdle");
 		session.closeNow();
     }
 package com.bsth.data.gpsdata.client.pf.codec;
+import com.bsth.data.gpsdata.client.pd.common.ConvertUtil;
 import com.bsth.data.gpsdata.client.pf.protocol.PfMessage;
 import org.apache.mina.core.buffer.IoBuffer;
 import org.apache.mina.core.session.IoSession;
 import org.apache.mina.filter.codec.CumulativeProtocolDecoder;
 import org.apache.mina.filter.codec.ProtocolDecoderOutput;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 public class MessageDecoder extends CumulativeProtocolDecoder {
+	Logger log = LoggerFactory.getLogger(MessageDecoder.class);
+
 	@Override
 	protected boolean doDecode(IoSession session, IoBuffer in,
 			ProtocolDecoderOutput out) throws Exception {
@@ -19,11 +24,18 @@ public class MessageDecoder extends CumulativeProtocolDecoder {
 			int len = ((lenh & 0xff) << 8) + (lenl & 0xff);
 			if ((head1 & 0xff) == 0xfa && (head2 & 0xff) == 0xf5) {
 				if (in.remaining() > len) {
-					byte[] bytes = new byte[len + 1];
-					in.get(bytes);
-					PfMessage msg = new PfMessage();
-					msg.read(bytes);
-					out.write(msg);
+					try {
+						byte[] bytes = new byte[len + 1];
+						in.get(bytes);
+						PfMessage msg = new PfMessage();
+						msg.read(bytes);
+
+						//日志纪录
+						log.info("pf client receive: " + ConvertUtil.bytesToHexString(bytes));
+						out.write(msg);
+					}catch (Exception e){
+						log.error("pf message decoder：", e);
+					}
 				} else {
 					in.reset();
 					return false;
@@ -49,6 +49,7 @@ public class PfClientHandler extends IoHandlerAdapter{
 	@Override
     public void sessionIdle(IoSession session, IdleStatus status) throws Exception {
+		log.warn("pf sessionIdle");
 		session.closeNow();
     }