add bingshui

src/main/java/com/tidecloud/dataacceptance/common/CRCUtil.java

@@ -0,0 +1,66 @@
+package com.tidecloud.dataacceptance.common;
+
+import java.math.BigDecimal;
+import java.text.DecimalFormat;
+import java.text.NumberFormat;
+
+/**   
+ * @author: chudk 
+ * @date:   2017年10月10日 下午3:21:59   
+ */
+public class CRCUtil {
+    
+    private static int crc16_ccitt_table[] = { 0X0000, 0X1189, 0X2312, 0X329B,
+            0X4624, 0X57AD, 0X6536, 0X74BF,
+            0X8C48, 0X9DC1, 0XAF5A, 0XBED3, 0XCA6C, 0XDBE5, 0XE97E, 0XF8F7,
+            0X1081, 0X0108, 0X3393, 0X221A, 0X56A5, 0X472C, 0X75B7, 0X643E,
+            0X9CC9, 0X8D40, 0XBFDB, 0XAE52, 0XDAED, 0XCB64, 0XF9FF, 0XE876,
+            0X2102, 0X308B, 0X0210, 0X1399, 0X6726, 0X76AF, 0X4434, 0X55BD,
+            0XAD4A, 0XBCC3, 0X8E58, 0X9FD1, 0XEB6E, 0XFAE7, 0XC87C, 0XD9F5,
+            0X3183, 0X200A, 0X1291, 0X0318, 0X77A7, 0X662E, 0X54B5, 0X453C,
+            0XBDCB, 0XAC42, 0X9ED9, 0X8F50, 0XFBEF, 0XEA66, 0XD8FD, 0XC974,
+            0X4204, 0X538D, 0X6116, 0X709F, 0X0420, 0X15A9, 0X2732, 0X36BB,
+            0XCE4C, 0XDFC5, 0XED5E, 0XFCD7, 0X8868, 0X99E1, 0XAB7A, 0XBAF3,
+            0X5285, 0X430C, 0X7197, 0X601E, 0X14A1, 0X0528, 0X37B3, 0X263A,
+            0XDECD, 0XCF44, 0XFDDF, 0XEC56, 0X98E9, 0X8960, 0XBBFB, 0XAA72,
+            0X6306, 0X728F, 0X4014, 0X519D, 0X2522, 0X34AB, 0X0630, 0X17B9,
+            0XEF4E, 0XFEC7, 0XCC5C, 0XDDD5, 0XA96A, 0XB8E3, 0X8A78, 0X9BF1,
+            0X7387, 0X620E, 0X5095, 0X411C, 0X35A3, 0X242A, 0X16B1, 0X0738,
+            0XFFCF, 0XEE46, 0XDCDD, 0XCD54, 0XB9EB, 0XA862, 0X9AF9, 0X8B70,
+            0X8408, 0X9581, 0XA71A, 0XB693, 0XC22C, 0XD3A5, 0XE13E, 0XF0B7,
+            0X0840, 0X19C9, 0X2B52, 0X3ADB, 0X4E64, 0X5FED, 0X6D76, 0X7CFF,
+            0X9489, 0X8500, 0XB79B, 0XA612, 0XD2AD, 0XC324, 0XF1BF, 0XE036,
+            0X18C1, 0X0948, 0X3BD3, 0X2A5A, 0X5EE5, 0X4F6C, 0X7DF7, 0X6C7E,
+            0XA50A, 0XB483, 0X8618, 0X9791, 0XE32E, 0XF2A7, 0XC03C, 0XD1B5,
+            0X2942, 0X38CB, 0X0A50, 0X1BD9, 0X6F66, 0X7EEF, 0X4C74, 0X5DFD,
+            0XB58B, 0XA402, 0X9699, 0X8710, 0XF3AF, 0XE226, 0XD0BD, 0XC134,
+            0X39C3, 0X284A, 0X1AD1, 0X0B58, 0X7FE7, 0X6E6E, 0X5CF5, 0X4D7C,
+            0XC60C, 0XD785, 0XE51E, 0XF497, 0X8028, 0X91A1, 0XA33A, 0XB2B3,
+            0X4A44, 0X5BCD, 0X6956, 0X78DF, 0X0C60, 0X1DE9, 0X2F72, 0X3EFB,
+            0XD68D, 0XC704, 0XF59F, 0XE416, 0X90A9, 0X8120, 0XB3BB, 0XA232,
+            0X5AC5, 0X4B4C, 0X79D7, 0X685E, 0X1CE1, 0X0D68, 0X3FF3, 0X2E7A,
+            0XE70E, 0XF687, 0XC41C, 0XD595, 0XA12A, 0XB0A3, 0X8238, 0X93B1,
+            0X6B46, 0X7ACF, 0X4854, 0X59DD, 0X2D62, 0X3CEB, 0X0E70, 0X1FF9,
+            0XF78F, 0XE606, 0XD49D, 0XC514, 0XB1AB, 0XA022, 0X92B9, 0X8330,
+            0X7BC7, 0X6A4E, 0X58D5, 0X495C, 0X3DE3, 0X2C6A, 0X1EF1, 0X0F78 };
+  
+    /** 
+     * 计算CRC16校验 
+     *  
+     * @param data 
+     *            需要计算的数组 
+     * @return CRC16校验值 
+     */  
+  
+    public static int do_crc(int regInit, byte[] message) {  
+        int crcReg = regInit;  
+        for (int i = 0; i < message.length; i++) {  
+            crcReg = (crcReg >> 8) ^ crc16_ccitt_table[(crcReg ^ message[i]) & 0xff];  
+        }  
+        return negate(~crcReg);
+    } 
+    
+    private static int negate(int data){
+        return data & 0x0FFFF;
+    }
+}

src/main/java/com/tidecloud/dataacceptance/common/DateUtil.java

@@ -1,5 +1,6 @@
 package com.tidecloud.dataacceptance.common;
 
+import java.text.SimpleDateFormat;
 import java.util.Date;
 import java.util.GregorianCalendar;
 
@@ -18,11 +19,20 @@ public class DateUtil {
     }
     
     private static Date calculate(Date d, int field, int amount) {   
-        if (d == null)   
-            return null;   
+        if (d == null) {
+            return null;
+        }
         GregorianCalendar g = new GregorianCalendar();   
         g.setTime(d);
         g.add(field, amount);   
         return g.getTime();   
     }  
+    
+    public static String formatDate2String(Date date){
+        if (date == null) {
+            return null;
+        }
+        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyMMddHHmmss");
+        return simpleDateFormat.format(date);
+    }
 }

src/main/java/com/tidecloud/dataacceptance/common/NumUtil.java

@@ -0,0 +1,62 @@
+package com.tidecloud.dataacceptance.common;
+
+import java.text.DecimalFormat;
+
+/**   
+ * @author: chudk 
+ * @date:   2017年10月16日 下午4:30:53   
+ */
+public class NumUtil {
+    
+    public static Double toFixed2Place(Double num){
+        if (num == null) {
+            return null;
+        }
+        DecimalFormat df = new DecimalFormat("######0.0");
+        return Double.valueOf(df.format(num));
+    }
+    
+    public static void main(String[] args) {
+        Double fixed2Place = NumUtil.toFixed2Place(1112.292d);
+        System.out.println(fixed2Place);
+    }
+    
+   public static byte[] int2bytes(int num)
+    {
+           byte[] b=new byte[4];
+           //int mask=0xff;
+           for(int i=0;i<4;i++){
+                b[3-i]=(byte)(num>>>(24-i*8));
+           }
+          return b;
+    }
+   
+   public static byte[] intToByte(int data, int length){
+       byte[] bytes = new byte[length];
+       bytes[0] = (byte)(data >> 8);
+       bytes[1] = (byte)(data & 0xff);
+       return bytes;
+   }
+   
+   public static String byte2String(byte b) {
+       if (b < 10) {
+           return "0" + b;
+       } else {
+           return String.valueOf(b);
+       }
+   }
+   
+   /**
+    * 把一个3位的数组转化位整形
+    * 
+    * @param value
+    * @return
+    * @throws Exception
+    */
+   public static int threeBytesToInteger(byte[] value) {
+       int temp0 = value[0] & 0xFF;
+       int temp1 = value[1] & 0xFF;
+       int temp2 = value[2] & 0xFF;
+       return ((temp0 << 16) + (temp1 << 8) + temp2);
+   }
+}

src/main/java/com/tidecloud/dataacceptance/entity/ConnectMsg.java

@@ -33,7 +33,7 @@ public class ConnectMsg implements Serializable {
      * @param ipAddr
      * @return int
      */
-    public String ipToLong(String ipAddress) {
+    public static String ipToLong(String ipAddress) {
         long result = 0;
         String[] ipAddressInArray = ipAddress.split("\\.");
         for (int i = 3; i >= 0; i--) {

src/main/java/com/tidecloud/dataacceptance/entity/YiTongGPSDevice.java

@@ -0,0 +1,106 @@
+package com.tidecloud.dataacceptance.entity;
+
+import java.io.Serializable;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+
+import com.tidecloud.dataacceptance.common.DateUtil;
+
+/**   
+ * @author: chudk 
+ * @date:   2017年9月18日 上午11:04:09   
+ */
+public class YiTongGPSDevice implements Serializable{
+
+    private static final long serialVersionUID = 684382625782853174L;
+    
+    private String deviceId;
+    private String date;
+    private String lat;
+    private String lng;
+    private String speed;
+    private String mileage;
+    
+    private Integer oil;
+    
+   
+    public Integer getOil() {
+        return oil;
+    }
+    public void setOil(Integer oil) {
+        this.oil = oil;
+    }
+    public YiTongGPSDevice() {
+    }
+    public String getDeviceId() {
+        return deviceId;
+    }
+
+    public void setDeviceId(String deviceId) {
+        this.deviceId = deviceId;
+    }
+
+    public String getDate() {
+        return date;
+    }
+
+    public void setDate(String date) {
+        this.date = date;
+    }
+
+    public String getSpeed() {
+        return speed;
+    }
+
+    public void setSpeed(String speed) {
+        this.speed = speed;
+    }
+
+    public String getLat() {
+        return lat;
+    }
+    public void setLat(String lat) {
+        this.lat = lat;
+    }
+    public String getLng() {
+        return lng;
+    }
+    public void setLng(String lng) {
+        this.lng = lng;
+    }
+    public String getMileage() {
+        return mileage;
+    }
+    public void setMileage(String mileage) {
+        this.mileage = mileage;
+    }
+    public static String buildDeviceStr(YiTongGPSDevice device){
+        // [设备号] [日期] [gpsCount] [经度] [维度] [速度] [里程] [是否补传] [数据上报模式] [移动基站] [LAC] [MCC] [信息类型 1代表location， 0代表电压] [电压]
+        SimpleDateFormat sdf = new SimpleDateFormat("yyMMddHHmmss");
+        Date date = new Date();
+        try {
+            date = sdf.parse(device.getDate());
+        } catch (ParseException e) {
+            e.printStackTrace();
+        }
+        date = DateUtil.calculateByHour(date, 8);
+        String dateStr = sdf.format(date);
+        StringBuilder builder = new StringBuilder();
+        builder.append(checkoutNullObject(device.getDeviceId())).append(",");
+        builder.append(checkoutNullObject(dateStr)).append(",");
+        builder.append(checkoutNullObject(device.getLng())).append(",");
+        builder.append(checkoutNullObject(device.getLat())).append(",");
+        builder.append(checkoutNullObject(device.getSpeed())).append(",");
+        builder.append(checkoutNullObject(device.getMileage())).append(",");
+        builder.append(checkoutNullObject(device.getOil()));
+        return builder.toString();
+    }
+    
+    private static String checkoutNullObject(Object t){
+        if (t == null) {
+            return "";
+        }
+        return String.valueOf(t);
+    }
+}

src/main/java/com/tidecloud/dataacceptance/service/AcceptanceDeviceService.java

@@ -8,6 +8,7 @@ import java.text.ParseException;
 import java.util.ArrayList;
 import java.util.Enumeration;
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -25,8 +26,11 @@ import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.stereotype.Service;
 
+import com.smartsanitation.common.util.StringUtil;
 import com.tidecloud.dataacceptance.config.AcceptanceDeviceConfig;
 import com.tidecloud.dataacceptance.config.DeviceConfig;
+import com.tidecloud.dataacceptance.entity.ConnectMsg;
+import com.tidecloud.dataacceptance.service.impl.YiTongGpsServerHandler;
 
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.ChannelFuture;
@@ -37,6 +41,8 @@ import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.handler.timeout.IdleStateHandler;
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.JedisPool;
 
 /**
  * Created by ryan
@@ -49,8 +55,11 @@ public class AcceptanceDeviceService implements ApplicationContextAware {
 	@Autowired
 	private AcceptanceDeviceConfig acceptanceDeviceListConfig;
 
+	@Autowired
+	private JedisPool jedisPool;
+
 	@PostConstruct
-	public void transfer() throws IOException, ParseException {
+	public void acceptance() throws IOException, ParseException {
 
 		List<DeviceConfig> deviceListConfig = acceptanceDeviceListConfig.getDeviceList();
 		List<DeviceConfig> workDeviceList = new ArrayList<>();
@@ -107,7 +116,7 @@ public class AcceptanceDeviceService implements ApplicationContextAware {
 							if (ip.getHostAddress().equals("127.0.0.1")) {
 								continue;
 							}
-							return	hostAddress = ip.getHostAddress();
+							return hostAddress = ip.getHostAddress();
 						}
 					}
 				}
@@ -119,6 +128,7 @@ public class AcceptanceDeviceService implements ApplicationContextAware {
 	}
 
 	public void start(AcceptanceInboundHandlerAdapter handler) {
+		cleanRedisLinkData(handler.getIp());
 		EventLoopGroup bossGroup = new NioEventLoopGroup();
 		EventLoopGroup workerGroup = new NioEventLoopGroup();
 		try {
@@ -139,12 +149,40 @@ public class AcceptanceDeviceService implements ApplicationContextAware {
 		} catch (Exception e) {
 			logger.error(e.getMessage());
 		} finally {
+			cleanRedisLinkData(handler.getIp());
 			workerGroup.shutdownGracefully();
 			bossGroup.shutdownGracefully();
 		}
 
 	}
 
+	private void cleanRedisLinkData(String address) {
+
+		try (Jedis jedis = jedisPool.getResource()) {
+			jedis.select(YiTongGpsServerHandler.REDIS_INDEX_LINK);
+			String addressStr = ConnectMsg.ipToLong(address);
+			String selectKey = YiTongGpsServerHandler.PREFIX_LINK_BACK + addressStr;
+			Set<String> values = jedis.smembers(selectKey);
+
+			for (String deviceId : values) {
+				String deleteKeyOfDevice = YiTongGpsServerHandler.PREFIX_DEVICE + deviceId;
+				String deleteKeyOfLink = YiTongGpsServerHandler.PREFIX_LINK + addressStr;
+				String connectMsgStr = jedis.get(deleteKeyOfDevice);
+				if (connectMsgStr != null) {
+					ConnectMsg connectMsg = StringUtil.convert2Object(connectMsgStr, ConnectMsg.class);
+					String socketId = connectMsg.getSocketId();
+					jedis.del(deleteKeyOfDevice);
+					jedis.srem(deleteKeyOfLink, socketId);
+				} else {
+					logger.error("error deviceId [{}] in select [{}] key [{}]", deviceId, 15, deleteKeyOfDevice);
+				}
+			}
+			jedis.del(selectKey);
+		} catch (Exception e) {
+			logger.error(e.getLocalizedMessage());
+		}
+	}
+
 	@Override
 	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
 		this.context = applicationContext;

src/main/java/com/tidecloud/dataacceptance/service/AcceptanceInboundHandlerAdapter.java

@@ -2,8 +2,10 @@ package com.tidecloud.dataacceptance.service;
 
 import java.io.BufferedWriter;
 import java.io.File;
+import java.io.FileOutputStream;
 import java.io.FileWriter;
 import java.io.IOException;
+import java.io.OutputStream;
 import java.nio.ByteBuffer;
 import java.nio.CharBuffer;
 import java.nio.charset.Charset;
@@ -16,6 +18,7 @@ import java.util.UUID;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
+import org.apache.commons.lang.ArrayUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -26,6 +29,7 @@ import org.springframework.util.concurrent.ListenableFuture;
 import org.springframework.util.concurrent.SuccessCallback;
 
 import com.smartsanitation.common.util.StringUtil;
+import com.tidecloud.dataacceptance.common.NumUtil;
 import com.tidecloud.dataacceptance.entity.ConnectMsg;
 
 import io.netty.buffer.ByteBuf;
@@ -36,25 +40,32 @@ import redis.clients.jedis.Jedis;
 import redis.clients.jedis.JedisPool;
 
 public abstract class AcceptanceInboundHandlerAdapter extends ChannelInboundHandlerAdapter {
-	
+
 	protected String ip;
-	
+
 	protected Integer port;
 
 	protected String dataPath;
 
-	protected String prefixName ;
-	
+	protected String prefixName;
+
 	private static final Logger logger = LoggerFactory.getLogger(AcceptanceInboundHandlerAdapter.class);
 	private static final Long TEN_M = 10485760l;
-	
-	private static String PREFIX_LINK = "s.";
-	private static String PREFIX_DEVICE = "d.";
+
+	public static String PREFIX_LINK = "s.";
+	public static String PREFIX_LINK_BACK = "s.b.";
+	public static String PREFIX_DEVICE = "d.";
+
+	public static final Integer REDIS_INDEX_LINK = 15;
 
 	public static Map<String, String> channelMap = new HashMap<String, String>();
 	public static Map<String, Channel> manageChannelMap = new HashMap<>();
 	public static Map<String, Channel> channelMapOfChannelKey = new HashMap<>();
 	public static Map<String, String> commandCopy = new HashMap<>();
+
+	public static Map<String, Channel> socketyChannelMap = new HashMap<>();
+	public static Map<Channel, String> channelDeviceMap = new HashMap<>();
+
 	private static ExecutorService singleThreadPool = Executors.newSingleThreadExecutor();
 
 	private static ExecutorService kafkaSendThreadPool = Executors.newSingleThreadExecutor();
@@ -65,57 +76,56 @@ public abstract class AcceptanceInboundHandlerAdapter extends ChannelInboundHand
 
 	private String topic;
 
-
 	@Override
-    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
-        ByteBuf byteBuf = (ByteBuf)msg;
-        String str = byteBufferToString(byteBuf.nioBuffer());
-        logger.info("上传数据：{}", str);
-       
-        try {
-            reply(ctx, str);
-        } catch (Exception e) {
-            logger.error(e.getMessage());
-        } finally {
-        	byteBuf.release();
-        }
-    }
-	abstract public void reply(ChannelHandlerContext ctx, String msg) throws Exception;
-	
-	protected void sendMsg(String msg,String deviceId) {
-    	kafkaSendThreadPool.execute(()->sendKakfaMsg(msg,deviceId));
-        singleThreadPool.execute(()->dataStorage(msg));
-    }
+	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+		ByteBuf byteBuf = (ByteBuf) msg;
+		String str = byteBufferToString(byteBuf.nioBuffer());
+		logger.info("接入数据：{}", str);
 
+		try {
+			reply(ctx, str);
+		} catch (Exception e) {
+			logger.error(e.getMessage());
+		} finally {
+			byteBuf.release();
+		}
+	}
 
+	abstract public void reply(ChannelHandlerContext ctx, String msg) throws Exception;
+	
 
-	public void manageLink(Channel channel, String deviceId) {
-		String channelId = UUID.randomUUID().toString();
-		manageChannelMap.put(channelId, channel);
-		channelMap.put(channelId, deviceId);
-		channelMapOfChannelKey.put(deviceId, channel);
-		logger.info("链接管理,链接id [{}]", channelId);
+	protected void receiveMsg(String msg, String deviceId, Channel channel) {
+		Channel channelInMap = channelMapOfChannelKey.get(deviceId);
+		if (channelInMap == null) {
+			manageChannel(channel, deviceId);
+		}
+		kafkaSendThreadPool.execute(() -> sendKakfaMsg(msg, deviceId));
+		singleThreadPool.execute(() -> dataStorage(msg));
+	}
 
-		ConnectMsg cMsg = new ConnectMsg(this.ip, channelId);
+	private void manageChannel(Channel channel, String deviceId) {
+		String socketkey = UUID.randomUUID().toString();
+		socketyChannelMap.put(socketkey, channel);
+		channelDeviceMap.put(channel, deviceId);
+		String addressStr = ConnectMsg.ipToLong(this.ip);
+		ConnectMsg cMsg = new ConnectMsg(this.ip, socketkey);
 		try (Jedis jedis = jedisPool.getResource()) {
-			jedis.select(15);
-			String insertKey = PREFIX_LINK + ip;
+			jedis.select(REDIS_INDEX_LINK);
+			String insertKey = PREFIX_LINK + addressStr;
 			String selectKey = PREFIX_DEVICE + deviceId;
-
-			jedis.set(insertKey, deviceId);
+			String insertBackupKey = PREFIX_LINK_BACK + addressStr;
+			jedis.sadd(insertKey, socketkey);
+			jedis.sadd(insertBackupKey, deviceId);
 			jedis.set(selectKey, StringUtil.convert2String(cMsg));
 		} catch (Exception e) {
-			logger.error(e.getLocalizedMessage());
+			logger.error(e.getMessage(), e);
 		}
 	}
 
-
-
-
 	protected void sendKakfaMsg(String msg, String deviceId) {
 		Integer key = new Integer(deviceId.hashCode());
 		ListenableFuture<SendResult<Integer, String>> listenableFuture = kafkaTemplate.send(topic, key, msg);
-		
+
 		// 发送成功回调
 		SuccessCallback<SendResult<Integer, String>> successCallback = new SuccessCallback<SendResult<Integer, String>>() {
 			@Override
@@ -162,24 +172,37 @@ public abstract class AcceptanceInboundHandlerAdapter extends ChannelInboundHand
 		}
 	}
 
+	// public static void writeDevice2File(File file, String deviceStr) {
+	// BufferedWriter bufferedWriter = null;
+	// try {
+	// bufferedWriter = new BufferedWriter(new FileWriter(file, true));
+	// bufferedWriter.write(deviceStr);
+	// bufferedWriter.newLine();
+	// bufferedWriter.flush();
+	// } catch (IOException e) {
+	// logger.error(e.getMessage());
+	// } finally {
+	// if (bufferedWriter != null) {
+	// try {
+	// bufferedWriter.close();
+	// } catch (IOException e) {
+	// logger.error("close file[{}] failed: {}", file.getName(), e.getMessage());
+	// }
+	// }
+	//
+	// }
+	// }
+
 	public static void writeDevice2File(File file, String deviceStr) {
-		BufferedWriter bufferedWriter = null;
+		Integer length = deviceStr.getBytes().length;
 		try {
-			bufferedWriter = new BufferedWriter(new FileWriter(file, true));
-			bufferedWriter.write(deviceStr);
-			bufferedWriter.newLine();
-			bufferedWriter.flush();
+			OutputStream outputStream = new FileOutputStream(file, true);
+			byte[] outPutBytes = ArrayUtils.addAll(NumUtil.int2bytes(length), deviceStr.getBytes());
+			outputStream.write(outPutBytes);
+			outputStream.flush();
+			outputStream.close();
 		} catch (IOException e) {
 			logger.error(e.getMessage());
-		} finally {
-			if (bufferedWriter != null) {
-				try {
-					bufferedWriter.close();
-				} catch (IOException e) {
-					logger.error("close file[{}] failed: {}", file.getName(), e.getMessage());
-				}
-			}
-
 		}
 	}
 
@@ -194,6 +217,39 @@ public abstract class AcceptanceInboundHandlerAdapter extends ChannelInboundHand
 		saveChannel(ctx);
 	}
 
+	@Override
+	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+		Channel channel = ctx.channel();
+		if (!channel.isActive()) {
+			String deviceId = channelDeviceMap.get(channel);
+			if (deviceId != null) {
+				channelDeviceMap.remove(channel);
+				// deleteLinkFromRedis(deviceId);
+			}
+		}
+		super.channelInactive(ctx);
+		ctx.close();
+	}
+
+	private void deleteLinkFromRedis(String deviceId) {
+		String deleteKey = PREFIX_DEVICE + deviceId;
+		try (Jedis jedis = jedisPool.getResource()) {
+			jedis.select(REDIS_INDEX_LINK);
+			String connectMsg = jedis.get(deleteKey);
+			if (connectMsg != null) {
+				ConnectMsg cmsg = StringUtil.convert2Object(connectMsg, ConnectMsg.class);
+				String socketId = cmsg.getSocketId();
+				socketyChannelMap.remove(socketId);
+				String socketQueryKey = PREFIX_LINK + this.ip;
+				jedis.srem(socketQueryKey, socketId);
+				jedis.del(deleteKey);
+				logger.info("delete link [{}] from redis and memory deviceId is [{}]", socketId, deviceId);
+			}
+		} catch (Exception e) {
+			logger.error(e.getLocalizedMessage());
+		}
+	}
+
 	private void saveChannel(ChannelHandlerContext ctx) {
 	}
 
@@ -249,21 +305,27 @@ public abstract class AcceptanceInboundHandlerAdapter extends ChannelInboundHand
 	public void setTopic(String topic) {
 		this.topic = topic;
 	}
+
 	public String getIp() {
 		return ip;
 	}
+
 	public void setIp(String ip) {
 		this.ip = ip;
 	}
+
 	public Integer getPort() {
 		return port;
 	}
+
 	public void setPort(Integer port) {
 		this.port = port;
 	}
+
 	public String getPrefixName() {
 		return prefixName;
 	}
+
 	public void setPrefixName(String prefixName) {
 		this.prefixName = prefixName;
 	}

src/main/java/com/tidecloud/dataacceptance/service/impl/WatchServerHandler.java

@@ -30,25 +30,20 @@ public class WatchServerHandler extends AcceptanceInboundHandlerAdapter {
 	public static final String name = "WatchServerHandler";
     private static final Logger logger = LoggerFactory.getLogger(WatchServerHandler.class);
    
-    public WatchServerHandler() {
-    	System.out.println(name);
-    }
+  
     public void reply(ChannelHandlerContext ctx, String msg) throws Exception {
         logger.info("设备上传数据：" + msg);
         Advice advice = getAdevice(msg);
         String deviceId = advice.getDeviceId();
         String adviceType = advice.getAdviceType();
         Channel channel = ctx.channel();
-        Channel channelInMap = channelMapOfChannelKey.get(deviceId);
-        if (channelInMap == null) {
-            manageLink(channel, deviceId);
-        }
+      
         switch (adviceType) {
         
         case "UD":
         case "heart":	
         case "UD2":
-        	sendMsg(msg,deviceId);
+        	receiveMsg(msg,deviceId,channel);
             logger.info("正常存储设备信息：" + getDevice(msg).toString());
             break;
         case "LK":
@@ -63,11 +58,6 @@ public class WatchServerHandler extends AcceptanceInboundHandlerAdapter {
         }
     }
     
-   
-    
- 
-
-   
 
     private void normalReply(Advice advice, Channel channel) {
         String facotry = advice.getFacotry();
@@ -187,6 +177,7 @@ public class WatchServerHandler extends AcceptanceInboundHandlerAdapter {
 //            new WatchServerHandler().dataStorage(device);
 //        }
     }
+	
 
    
 }

src/main/java/com/tidecloud/dataacceptance/service/impl/YiTongGpsServerHandler.java

@@ -0,0 +1,78 @@
+package com.tidecloud.dataacceptance.service.impl;
+
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+import com.tidecloud.dataacceptance.service.AcceptanceInboundHandlerAdapter;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.GenericFutureListener;
+
+/**
+ * @author cdk
+ */
+@Component
+@ChannelHandler.Sharable
+public class YiTongGpsServerHandler extends AcceptanceInboundHandlerAdapter {
+
+	private static final Logger logger = LoggerFactory.getLogger(YiTongGpsServerHandler.class);
+
+	public void reply(ChannelHandlerContext ctx, String oringalData) {
+		Map<String, String> dataMap = new HashMap<>();
+
+		String[] dataArray = oringalData.split("\\s+");
+		String phone = dataArray[0];
+		String deviceData = dataArray[1];
+		if (deviceData.endsWith("]") && deviceData.startsWith("[")) {
+			String subDeviceData = deviceData.substring(1, deviceData.length() - 1);
+			String[] subDataArray = subDeviceData.split(";");
+			try {
+				for (int i = 0; i < subDataArray.length; i++) {
+					String subData = subDataArray[i];
+					String[] datas = subData.split(":");
+					if (datas[0].contains("S") && !datas[0].contains("G")) {
+						dataMap.put("S", datas[1]);
+					} else {
+						dataMap.put(datas[0], datas[1]);
+					}
+				}
+			} catch (Exception e) {
+				logger.error(e.getMessage(), e);
+			}
+		}
+		Channel channel = ctx.channel();
+		
+		receiveMsg(oringalData,phone,channel);
+
+
+		String serialStr = dataMap.get("S");
+		Integer serialId = Integer.valueOf(serialStr) + 1;
+		Date now = new Date();
+		SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyyMMddHHmmss");
+		String dataStr = simpleDateFormat.format(now);
+		String replyMsg = "[" + "S:" + serialId + ";R" + "1" + ";" + "ST:" + dataStr + "]";
+		ByteBuf buffer = Unpooled.buffer(replyMsg.getBytes().length);
+		buffer.writeBytes(replyMsg.getBytes());
+
+		ChannelFuture channelFuture = channel.writeAndFlush(buffer);
+		channelFuture.addListener(new GenericFutureListener<Future<? super Void>>() {
+			@Override
+			public void operationComplete(Future<? super Void> future) throws Exception {
+				logger.info("回复设备: [{}] 成功", replyMsg);
+			}
+		});
+	}
+
+}

src/main/java/com/tidecloud/dataacceptance/web/DeviceController.java

@@ -0,0 +1,107 @@
+package com.tidecloud.dataacceptance.web;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.commons.lang.ArrayUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import com.smartsanitation.common.response.ResultWrapper;
+import com.smartsanitation.common.util.StringUtil;
+import com.tidecloud.dataacceptance.common.CRCUtil;
+import com.tidecloud.dataacceptance.common.NumUtil;
+import com.tidecloud.dataacceptance.entity.ConnectMsg;
+import com.tidecloud.dataacceptance.service.impl.YiTongGpsServerHandler;
+
+import io.netty.channel.Channel;
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.JedisPool;
+
+/**   
+ * @author: chudk 
+ * @date:   2017年10月30日 上午9:44:50   
+ */
+@RestController("setting")
+public class DeviceController {
+    
+    private static final Logger logger = LoggerFactory.getLogger(DeviceController.class);
+    
+    @Autowired
+    private JedisPool jedisPool;
+    
+    @RequestMapping("cuttingOilAndElecticity")
+    public ResultWrapper<?> deviceController(@RequestParam String deviceIds, @RequestParam String commandStr){
+        Map<String, ConnectMsg> deviceIdSocketIdMap = getDeviceSocketId(deviceIds);
+        if (deviceIdSocketIdMap.isEmpty()) {
+            return ResultWrapper.ok();
+        }
+        try {
+            Map<String, Channel> channelMap = YiTongGpsServerHandler.socketyChannelMap;
+            if (channelMap.isEmpty()) {
+                return ResultWrapper.ok();
+            }
+            for (Map.Entry<String, ConnectMsg> deviceIdSocketIdEntry : deviceIdSocketIdMap.entrySet()) {
+                ConnectMsg connectMsg = deviceIdSocketIdEntry.getValue();
+//                String deviceId = deviceIdSocketIdEntry.getKey();
+                Channel channel = channelMap.get(connectMsg.getSocketId());
+                if (channel != null && channel.isActive()) {
+                    logger.info("start write command to device the address [{}]", channel.remoteAddress());
+                    sendCommand(channel, commandStr, (byte)0x80);
+                    logger.info("end write command to device the address [{}]", channel.remoteAddress());
+                }
+            }
+        } catch (Exception e) {
+            logger.error(e.getLocalizedMessage());
+        }
+        return null;
+    }
+
+    private void sendCommand(Channel channel, String commandStr, byte protocalNum) {
+        byte commandLength = (byte) (4 + commandStr.getBytes().length);
+        byte dataPackageLength = (byte) ( 8 + commandLength);
+        
+        byte[] subCommandBytes = new byte[]{dataPackageLength, protocalNum, 
+                commandLength, 0x00, 0x00, 0x00, 0x00};
+        byte[] commandBytes = ArrayUtils.addAll(subCommandBytes, commandStr.getBytes());
+        
+        byte[] commandEnding = new byte[]{0x00, 0x01};
+        byte[] command = ArrayUtils.addAll(commandBytes, commandEnding);
+        int doCrc = CRCUtil.do_crc(65535, command);
+        
+        byte[] intToByte = NumUtil.intToByte(doCrc, 2);
+    
+        byte[] startBytes = new byte[]{0x78, 0x78};
+        byte[] endBytes = new byte[]{0x0d, 0x0a};
+        byte[] startCommandBytes = ArrayUtils.addAll(startBytes, command);
+        byte[] lastCommandBytes = ArrayUtils.addAll(startCommandBytes, intToByte);
+        byte[] lastCommand = ArrayUtils.addAll(lastCommandBytes, endBytes);
+        
+        channel.writeAndFlush(lastCommand);
+    }
+
+    private Map<String, ConnectMsg> getDeviceSocketId(String deviceIds) {
+        Map<String, ConnectMsg> deviceIdSocketIdMap = new HashMap<String, ConnectMsg>();
+        String[] deviceIdsArray = deviceIds.split(",");
+        try (Jedis jedis = jedisPool.getResource()) {
+            jedis.select(YiTongGpsServerHandler.REDIS_INDEX_LINK);
+            for (String deviceId : deviceIdsArray) {
+                String selectKey = YiTongGpsServerHandler.PREFIX_DEVICE + deviceId;
+                String result = jedis.get(selectKey);
+                if (result != null) {
+                    ConnectMsg connectMsg = StringUtil.convert2Object(result, ConnectMsg.class);
+                    deviceIdSocketIdMap.put(deviceId, connectMsg);
+                }
+            }
+        } catch (Exception e) {
+            logger.error(e.getMessage(), e);
+        } finally {
+            jedisPool.close();
+        }
+        return deviceIdSocketIdMap;
+    }
+}

src/main/resources/application.yml

@@ -4,10 +4,14 @@ spring:
   kafka:
     bootstrap-servers: 172.19.0.24:9092
   redis:
-    host: 192.168.0.119
-    port: 6379
-    timeout: 10000
+    host: 10.25.48.128
     password: tidecloudredis
+    timeout: 10000
+    port: 6379
+    max: 100
+    maxIdle: 10
+    minIdle: 3
+    maxWaitMills: 10000
 eureka:
   client:
     service-url:
@@ -28,7 +32,15 @@ acceptance:
     dataFileDir: /home/service/collector_watch/rawdata/
     handlerClass: com.tidecloud.dataacceptance.service.impl.WatchServerHandler
     enable: true
-
+    
+   -
+    name: bingshui
+    topic: device-yitong-bingshui
+    ip: 10.27.118.76
+    port: 7510
+    dataFileDir: /home/service/collector_7510/rawdata/
+    handlerClass: com.tidecloud.dataacceptance.service.impl.YiTongGpsServerHandler
+    enable: true
       
 logging:
   config: