update...

潘钊
1 parent bed43f38
Showing 9 changed files with 195 additions and 109 deletions
src/main/java/com/bsth/client/ClientApp.java
src/main/java/com/bsth/client/GpsBeforeBuffer.java
src/main/java/com/bsth/client/pd/codec/MessageDecoder.java
src/main/java/com/bsth/client/pd/handler/PdClientHandler.java
src/main/java/com/bsth/client/pf/codec/MessageDecoder.java
src/main/java/com/bsth/client/pf/handler/PfClientHandler.java
src/main/java/com/bsth/controller/GpsRealDataController.java
src/main/java/com/bsth/entity/GpsEntity.java
src/main/resources/logback.xml
@@ -18,6 +18,8 @@ import org.apache.mina.transport.socket.nio.NioSocketConnector;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.core.annotation.Order;
 import org.springframework.stereotype.Component;
 import java.net.InetSocketAddress;
@@ -27,7 +29,8 @@ import java.util.concurrent.*;
  * Created by panzhao on 2017/5/4.
  */
 @Component
-public class ClientApp {
+@Order(1)
+public class ClientApp implements CommandLineRunner{
     private static NioSocketConnector pdDataConnector;
     private static NioSocketConnector pfDataConnector;
@@ -67,7 +70,6 @@ public class ClientApp {
             logger.info("dconnect...");
             pdSession = session;
-            //SessionManager.getInstance().register(device, session);
         } catch (Exception e) {
             e.printStackTrace();
         }
@@ -212,6 +214,12 @@ public class ClientApp {
     static IoSession pdSession;
     static IoSession pfSession;
+
+    @Override
+    public void run(String... strings) throws Exception {
+        init();
+    }
+
     final class SessionChecker implements Runnable {
         @Override
 package com.bsth.client;
-import com.bsth.client.pd.protocol.BasicInfo;
 import com.bsth.entity.GpsEntity;
 import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.CommandLineRunner;
+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;
 /**
- * 从 socket client 到 .. 的缓冲
+ * 从 socket client 到调度系统 的缓冲
  * Created by panzhao on 2017/5/4.
  */
 @Component
-public class GpsBeforeBuffer {
+@Order(2)
+public class GpsBeforeBuffer implements CommandLineRunner{
-    static LinkedList<GpsEntity> linkedList = new LinkedList<>();
+    static LinkedList<GpsEntity> linkedList = new LinkedList();
+    static final int MAX_SIZE = 4000 * 20;
-    public void put(BasicInfo basicInfo){
-        //放弃补发数据
-        byte cacheData = getCacheState(basicInfo.getServiceState());
-        if(cacheData == 1)
-            return;
-        GpsEntity gps = new GpsEntity();
-
-        gps.setDeviceId(basicInfo.getDeviceId());
-        gps.setTimestamp(basicInfo.getTimestamp());
-        gps.setLat(basicInfo.getLat());
-        gps.setLon(basicInfo.getLon());
-        gps.setDirection((float)basicInfo.getDirection() / 10);
-        gps.setValid(basicInfo.getGpsValid());
-        gps.setCompanyCode(basicInfo.getCompanyCode());
-        gps.setStopNo(basicInfo.getStopNo());
-        gps.setUpDown(basicInfo.getUpOrDown());
-        gps.setSpeed((float)basicInfo.getSpeedGps() / 10);
-        gps.setLineId(String.valueOf(basicInfo.getLineId()));
-        gps.setState((int) getService(basicInfo.getServiceState()));
-        //没有设备号
-        if (StringUtils.isBlank(gps.getDeviceId()))
+    public void put(GpsEntity gps){
+        if(gps == null || StringUtils.isBlank(gps.getDeviceId()))
             return;
-        linkedList.addLast(gps);
+        linkedList.add(gps);
     }
-    public byte getCacheState(long serviceState) {
-        return (byte)(((serviceState & 0x00100000) == 0x00100000) ? 1 : 0);
-    }
+    static Logger log = LoggerFactory.getLogger(GpsBeforeBuffer.class);
+    public static List<GpsEntity> pollAll(){
+        List<GpsEntity> rs = new ArrayList<>(300);
+        GpsEntity gps;
+        while (true){
+            gps = linkedList.poll();
+            if(gps == null){
+                break;
+            }
+            rs.add(gps);
+        }
+
+        log.info("poll size: " + rs.size());
+        return rs;
+    }
     /**
-     * 获取运营状态
-     *
-     * @return -1无效 0运营 1未运营
+     * 清理数据，保持最大 MAX_SIZE 个数的元素
      */
-    public static byte getService(long serviceState) {
-        if ((serviceState & 0x00020000) == 0x00020000 || (serviceState & 0x80000000) == 0x80000000)
-            return -1;
-        return (byte) (((serviceState & 0x02000000) == 0x02000000) ? 1 : 0);
+    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();
+        }
+        log.info("clear size: " + len);
+    }
+
+    @Autowired
+    BufferSizeCheck bufferSizeCheck;
+
+    ScheduledExecutorService sexec;
+
+    @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);
+    }
+
+    @Component
+    public static class BufferSizeCheck extends Thread {
+
+        @Override
+        public void run() {
+            GpsBeforeBuffer.clear();
+        }
     }
 }
 \ No newline at end of file
@@ -32,7 +32,7 @@ public class MessageDecoder extends CumulativeProtocolDecoder {
 						out.write(msg);
 						//日志纪录
-						log.info("pd client receive: " + ConvertUtil.bytesToHexString(bytes));
+						log.info("pd receive: " + ConvertUtil.bytesToHexString(bytes));
 					}catch (Exception e){
 						log.error("pd message decoder：", e);
 					}
@@ -5,6 +5,7 @@ import com.bsth.client.msg.IMessageBody;
 import com.bsth.client.pd.protocol.PdMessage;
 import com.bsth.client.pd.protocol.Pd_41_0;
 import com.bsth.client.pd.protocol.Pd_42_0;
+import com.bsth.entity.GpsEntity;
 import org.apache.mina.core.service.IoHandlerAdapter;
 import org.apache.mina.core.session.IdleStatus;
 import org.apache.mina.core.session.IoSession;
@@ -67,20 +68,16 @@ public class PdClientHandler extends IoHandlerAdapter{
         PdMessage msg = (PdMessage)message;
         IMessageBody body = msg.getMessageBody();
         if (body != null) {
-        	//String deviceId = body.getDeviceId();
         	if (0x31 == msg.getCommandType()) {
-        		//IoSession regSession = SessionManager.getInstance().getSession(deviceId);
-        		//if (regSession != null) regSession.closeNow();
-        		//SessionManager.getInstance().register(deviceId, session);
         		log.debug("设备编号:" + body.getDeviceId() + "建立连接");
         	}
             else if(0x41 == msg.getCommandType()){
                 Pd_41_0 pd41 = (Pd_41_0)msg.getMessageBody();
-                gpsBeforeBuffer.put(pd41.getInfo());
+                gpsBeforeBuffer.put(new GpsEntity(pd41.getInfo(), msg.getVersion(), 1));
             }
             else if(0x42 == msg.getCommandType()){
                 Pd_42_0 pd42 = (Pd_42_0)msg.getMessageBody();
-                gpsBeforeBuffer.put(pd42.getInfo());
+                gpsBeforeBuffer.put(new GpsEntity(pd42.getInfo(), msg.getVersion(), 1));
             }
         }
     }
@@ -31,7 +31,7 @@ public class MessageDecoder extends CumulativeProtocolDecoder {
 						msg.read(bytes);
 						//日志纪录
-						log.info("pf client receive: " + ConvertUtil.bytesToHexString(bytes));
+						log.info("pf receive: " + ConvertUtil.bytesToHexString(bytes));
 						out.write(msg);
 					}catch (Exception e){
 						log.error("pf message decoder：", e);
@@ -44,12 +44,4 @@ public class MessageDecoder extends CumulativeProtocolDecoder {
 		}
 		return false;
 	}
-
-	/*private static String toHexString(byte[] bytes) {
-		StringBuilder sb = new StringBuilder();
-		for (byte b : bytes) {
-			sb.append(Integer.toHexString(b & 0xff) + "|");
-		}
-		return sb.toString();
-	}*/
 }
@@ -5,6 +5,7 @@ import com.bsth.client.common.Protocol2BizUtil;
 import com.bsth.client.msg.IMessageBody;
 import com.bsth.client.pd.protocol.BasicInfo;
 import com.bsth.client.pf.protocol.PfMessage;
+import com.bsth.entity.GpsEntity;
 import org.apache.mina.core.service.IoHandlerAdapter;
 import org.apache.mina.core.session.IdleStatus;
 import org.apache.mina.core.session.IoSession;
@@ -70,7 +71,7 @@ public class PfClientHandler extends IoHandlerAdapter{
         	}
 			BasicInfo info = Protocol2BizUtil.getBasicInfoFromMsg(msg);
-			gpsBeforeBuffer.put(info);
+			gpsBeforeBuffer.put(new GpsEntity(info, msg.getVersion(), 0));
         }
     }
+package com.bsth.controller;
+
+import com.bsth.client.GpsBeforeBuffer;
+import com.bsth.entity.GpsEntity;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+
+/**
+ * Created by panzhao on 2017/5/12.
+ */
+@RestController
+@RequestMapping("realGps")
+public class GpsRealDataController {
+
+
+    @RequestMapping("/all")
+    public List<GpsEntity> all(){
+        return GpsBeforeBuffer.pollAll();
+    }
+}
 package com.bsth.entity;
+import com.bsth.client.pd.protocol.BasicInfo;
+
 /**
  * @author PanZhao
  * @ClassName: GpsRealData
@@ -24,26 +26,6 @@ public class GpsEntity {
     private String deviceId;
     /**
-     * 停车场编码
-     */
-    private String carparkNo;
-
-    /**
-     * 站点编码
-     */
-    private String stopNo;
-
-    /**
-     * 站点名称
-     */
-    private String stationName;
-
-    /**
-     * 到站时间
-     */
-    private long arrTime;
-
-    /**
      * 经度
      */
     private Float lon;
@@ -85,28 +67,55 @@ public class GpsEntity {
      */
     private int valid;
-    public String getDeviceId() {
-        return deviceId;
-    }
+    /**
+     * 数据来源
+     * 1：网关
+     * 0：转发
+     */
+    private int source;
-    public void setDeviceId(String deviceId) {
-        this.deviceId = deviceId;
+    public GpsEntity(BasicInfo basicInfo, int version, int source) {
+        //放弃补发数据
+        byte cacheData = getCacheState(basicInfo.getServiceState());
+        if(cacheData == 1)
+            return;
+
+        this.setDeviceId(basicInfo.getDeviceId());
+        this.setTimestamp(basicInfo.getTimestamp());
+        this.setLat(basicInfo.getLat());
+        this.setLon(basicInfo.getLon());
+        this.setDirection((float)basicInfo.getDirection() / 10);
+        this.setValid(basicInfo.getGpsValid());
+        this.setCompanyCode(basicInfo.getCompanyCode());
+        this.setUpDown(basicInfo.getUpOrDown());
+        this.setSpeed((float)basicInfo.getSpeedGps() / 10);
+        this.setLineId(String.valueOf(basicInfo.getLineId()));
+        this.setState((int) getService(basicInfo.getServiceState()));
+        this.setVersion(version);
+        this.setSource(source);
     }
-    public String getCarparkNo() {
-        return carparkNo;
+    public static byte getCacheState(long serviceState) {
+        return (byte)(((serviceState & 0x00100000) == 0x00100000) ? 1 : 0);
     }
-    public void setCarparkNo(String carparkNo) {
-        this.carparkNo = carparkNo;
+    /**
+     * 获取运营状态
+     *
+     * @return -1无效 0运营 1未运营
+     */
+    public static byte getService(long serviceState) {
+        if ((serviceState & 0x00020000) == 0x00020000 || (serviceState & 0x80000000) == 0x80000000)
+            return -1;
+        return (byte) (((serviceState & 0x02000000) == 0x02000000) ? 1 : 0);
     }
-    public String getStopNo() {
-        return stopNo;
+    public String getDeviceId() {
+        return deviceId;
     }
-    public void setStopNo(String stopNo) {
-        this.stopNo = stopNo;
+    public void setDeviceId(String deviceId) {
+        this.deviceId = deviceId;
     }
     public Float getLon() {
@@ -142,24 +151,6 @@ public class GpsEntity {
         this.state = state;
     }
-
-    public String getStationName() {
-        return stationName;
-    }
-
-    public void setStationName(String stationName) {
-        this.stationName = stationName;
-    }
-
-    public long getArrTime() {
-        return arrTime;
-    }
-
-    public void setArrTime(long arrTime) {
-        this.arrTime = arrTime;
-    }
-
-
     public String getLineId() {
         return lineId;
     }
@@ -218,4 +209,12 @@ public class GpsEntity {
     public void setSpeed(Float speed) {
         this.speed = speed;
     }
+
+    public int getSource() {
+        return source;
+    }
+
+    public void setSource(int source) {
+        this.source = source;
+    }
 }
@@ -9,7 +9,29 @@
 		<layout class="ch.qos.logback.classic.PatternLayout">
 			<!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度%msg：日志消息，%n是换行符 -->
-			<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] [%file:%line] %-5level-%msg%n
+			<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} -%msg%n
+			</pattern>
+		</layout>
+	</appender>
+
+	<!-- 浦东网关 -->
+	<appender name="FILE"
+			  class="ch.qos.logback.core.rolling.RollingFileAppender">
+		<file>${LOG_BASE}/main.log</file>
+		<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
+			<fileNamePattern>${LOG_BASE}/main-%d{yyyy-MM-dd}.%i.log</fileNamePattern>
+			<timeBasedFileNamingAndTriggeringPolicy
+					class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
+				<maxFileSize>100MB</maxFileSize>
+			</timeBasedFileNamingAndTriggeringPolicy>
+		</rollingPolicy>
+		<encoder>
+			<pattern>%msg%n</pattern>
+		</encoder>
+
+		<layout class="ch.qos.logback.classic.PatternLayout">
+			<!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度%msg：日志消息，%n是换行符 -->
+			<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} -%msg%n
 			</pattern>
 		</layout>
 	</appender>
@@ -31,10 +53,15 @@
 		<layout class="ch.qos.logback.classic.PatternLayout">
 			<!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度%msg：日志消息，%n是换行符 -->
-			<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] [%file:%line] %-5level-%msg%n
+			<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} -%msg%n
 			</pattern>
 		</layout>
 	</appender>
+	<logger name="com.bsth.client.pd.codec.MessageDecoder"
+			level="INFO" additivity="false">
+		<appender-ref ref="PD_FILE" />
+	</logger>
+
 	<!-- 浦东转发 -->
 	<appender name="PF_FILE"
@@ -53,10 +80,14 @@
 		<layout class="ch.qos.logback.classic.PatternLayout">
 			<!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度%msg：日志消息，%n是换行符 -->
-			<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] [%file:%line] %-5level-%msg%n
+			<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} %-5level-%msg%n
 			</pattern>
 		</layout>
 	</appender>
+	<logger name="com.bsth.client.pf.codec.MessageDecoder"
+			level="INFO" additivity="false">
+		<appender-ref ref="PF_FILE" />
+	</logger>
 	<!-- 日志输出级别 -->