关键词搜索

源码搜索 ×
×

Java解析JT808协议示例

发布2018-08-21浏览8928次

详情内容

原文地址：https://blog.csdn.net/hylexus/article/details/54987786

本篇文章将介绍JT808协议的解析思路。

1 JT808协议扫盲

1.1 数据类型

数据类型描述及要求
BYTE无符号单字节整形（字节， 8 位）
WORD无符号双字节整形（字， 16 位）
DWORD无符号四字节整形（双字， 32 位）
BYTE[n]n 字节
BCD[n]8421 码， n 字节
STRINGGBK 编码，若无数据，置空

1.2 消息结构

标识位消息头消息体校验码标识位
1byte(0x7e)16byte 1byte1byte(0x7e)

1.3 消息头

  1. 消息ID(0-1)   消息体属性(2-3)  终端手机号(4-9)  消息流水号(10-11)    消息包封装项(12-15)
  2.  
  3. byte[0-1]   消息ID word(16)
  4. byte[2-3]   消息体属性 word(16)
  5.         bit[0-9]    消息体长度
  6.         bit[10-12]  数据加密方式
  7.                         此三位都为 0，表示消息体不加密
  8.                         第 10 位为 1，表示消息体经过 RSA 算法加密
  9.                         其它保留
  10.         bit[13]     分包
  11.                         1：消息体卫长消息，进行分包发送处理，具体分包信息由消息包封装项决定
  12.                         0：则消息头中无消息包封装项字段
  13.         bit[14-15]  保留
  14. byte[4-9]   终端手机号或设备ID bcd[6]
  15.         根据安装后终端自身的手机号转换
  16.         手机号不足12 位，则在前面补 0
  17. byte[10-11]     消息流水号 word(16)
  18.         按发送顺序从 0 开始循环累加
  19. byte[12-15]     消息包封装项
  20.         byte[0-1]   消息包总数(word(16))
  21.                         该消息分包后得总包数
  22.         byte[2-3]   包序号(word(16))
  23.                         从 1 开始
  24.         如果消息体属性中相关标识位确定消息分包处理,则该项有内容
  25.         否则无该项

2 解析

整个消息体结构中最复杂的就是消息头了。

2.1 消息体实体类

以下是对整个消息体抽象出来的一个java实体类。

  1. import java.nio.channels.Channel;
  2.  
  3. public class PackageData {
  4.  
  5.     /**
  6.      * 16byte 消息头
  7.      */
  8.     protected MsgHeader msgHeader;
  9.  
  10.     // 消息体字节数组
  11.     protected byte[] msgBodyBytes;
  12.  
  13.     /**
  14.      * 校验码 1byte
  15.      */
  16.     protected int checkSum;
  17.  
  18.     //记录每个客户端的channel,以便下发信息给客户端
  19.     protected Channel channel;
  20.  
  21.     public MsgHeader getMsgHeader() {
  22.         return msgHeader;
  23.     }
  24.  
  25.     //TODO set 和 get 方法在此处省略
  26.  
  27.     //消息头
  28.     public static class MsgHeader {
  29.         // 消息ID
  30.         protected int msgId;
  31.  
  32.         /// ========消息体属性
  33.         // byte[2-3]
  34.         protected int msgBodyPropsField;
  35.         // 消息体长度
  36.         protected int msgBodyLength;
  37.         // 数据加密方式
  38.         protected int encryptionType;
  39.         // 是否分包,true==>有消息包封装项
  40.         protected boolean hasSubPackage;
  41.         // 保留位[14-15]
  42.         protected String reservedBit;
  43.         /// ========消息体属性
  44.  
  45.         // 终端手机号
  46.         protected String terminalPhone;
  47.         // 流水号
  48.         protected int flowId;
  49.  
  50.          =====消息包封装项
  51.         // byte[12-15]
  52.         protected int packageInfoField;
  53.         // 消息包总数(word(16))
  54.         protected long totalSubPackage;
  55.         // 包序号(word(16))这次发送的这个消息包是分包中的第几个消息包, 从 1 开始
  56.         protected long subPackageSeq;
  57.          =====消息包封装项
  58.  
  59.         //TODO set 和 get 方法在此处省略
  60.     }
  61.  
  62. }

2.2 字节数组到消息体实体类的转换

2.2.1 消息转换器

  1. import org.slf4j.Logger;
  2. import org.slf4j.LoggerFactory;
  3. import cn.hylexus.jt808.util.BCD8421Operater;
  4. import cn.hylexus.jt808.util.BitOperator;
  5. import cn.hylexus.jt808.vo.PackageData;
  6. import cn.hylexus.jt808.vo.PackageData.MsgHeader;
  7.  
  8. public class MsgDecoder {
  9.  
  10.     private static final Logger log = LoggerFactory.getLogger(MsgDecoder.class);
  11.  
  12.     private BitOperator bitOperator;
  13.     private BCD8421Operater bcd8421Operater;
  14.  
  15.     public MsgDecoder() {
  16.         this.bitOperator = new BitOperator();
  17.         this.bcd8421Operater = new BCD8421Operater();
  18.     }
  19.  
  20.     //字节数组到消息体实体类
  21.     public PackageData queueElement2PackageData(byte[] data) {
  22.         PackageData ret = new PackageData();
  23.  
  24.         // 1. 16byte 或 12byte 消息头
  25.         MsgHeader msgHeader = this.parseMsgHeaderFromBytes(data);
  26.         ret.setMsgHeader(msgHeader);
  27.  
  28.         int msgBodyByteStartIndex = 12;
  29.         // 2. 消息体
  30.         // 有子包信息,消息体起始字节后移四个字节:消息包总数(word(16))+包序号(word(16))
  31.         if (msgHeader.isHasSubPackage()) {
  32.             msgBodyByteStartIndex = 16;
  33.         }
  34.  
  35.         byte[] tmp = new byte[msgHeader.getMsgBodyLength()];
  36.         System.arraycopy(data, msgBodyByteStartIndex, tmp, 0, tmp.length);
  37.         ret.setMsgBodyBytes(tmp);
  38.  
  39.         // 3. 去掉分隔符之后，最后一位就是校验码
  40.         // int checkSumInPkg =
  41.         // this.bitOperator.oneByteToInteger(data[data.length - 1]);
  42.         int checkSumInPkg = data[data.length - 1];
  43.         int calculatedCheckSum = this.bitOperator.getCheckSum4JT808(data, 0, data.length - 1);
  44.         ret.setCheckSum(checkSumInPkg);
  45.         if (checkSumInPkg != calculatedCheckSum) {
  46.             log.warn("检验码不一致,msgid:{},pkg:{},calculated:{}", msgHeader.getMsgId(), checkSumInPkg, calculatedCheckSum);
  47.         }
  48.         return ret;
  49.     }
  50.  
  51.     private MsgHeader parseMsgHeaderFromBytes(byte[] data) {
  52.         MsgHeader msgHeader = new MsgHeader();
  53.  
  54.         // 1. 消息ID word(16)
  55.         // byte[] tmp = new byte[2];
  56.         // System.arraycopy(data, 0, tmp, 0, 2);
  57.         // msgHeader.setMsgId(this.bitOperator.twoBytesToInteger(tmp));
  58.         msgHeader.setMsgId(this.parseIntFromBytes(data, 0, 2));
  59.  
  60.         // 2. 消息体属性 word(16)=================>
  61.         // System.arraycopy(data, 2, tmp, 0, 2);
  62.         // int msgBodyProps = this.bitOperator.twoBytesToInteger(tmp);
  63.         int msgBodyProps = this.parseIntFromBytes(data, 2, 2);
  64.         msgHeader.setMsgBodyPropsField(msgBodyProps);
  65.         // [ 0-9 ] 0000,0011,1111,1111(3FF)(消息体长度)
  66.         msgHeader.setMsgBodyLength(msgBodyProps & 0x1ff);
  67.         // [10-12] 0001,1100,0000,0000(1C00)(加密类型)
  68.         msgHeader.setEncryptionType((msgBodyProps & 0xe00) >> 10);
  69.         // [ 13_ ] 0010,0000,0000,0000(2000)(是否有子包)
  70.         msgHeader.setHasSubPackage(((msgBodyProps & 0x2000) >> 13) == 1);
  71.         // [14-15] 1100,0000,0000,0000(C000)(保留位)
  72.         msgHeader.setReservedBit(((msgBodyProps & 0xc000) >> 14) + "");
  73.         // 消息体属性 word(16)<=================
  74.  
  75.         // 3. 终端手机号 bcd[6]
  76.         // tmp = new byte[6];
  77.         // System.arraycopy(data, 4, tmp, 0, 6);
  78.         // msgHeader.setTerminalPhone(this.bcd8421Operater.bcd2String(tmp));
  79.         msgHeader.setTerminalPhone(this.parseBcdStringFromBytes(data, 4, 6));
  80.  
  81.         // 4. 消息流水号 word(16) 按发送顺序从 0 开始循环累加
  82.         // tmp = new byte[2];
  83.         // System.arraycopy(data, 10, tmp, 0, 2);
  84.         // msgHeader.setFlowId(this.bitOperator.twoBytesToInteger(tmp));
  85.         msgHeader.setFlowId(this.parseIntFromBytes(data, 10, 2));
  86.  
  87.         // 5. 消息包封装项
  88.         // 有子包信息
  89.         if (msgHeader.isHasSubPackage()) {
  90.             // 消息包封装项字段
  91.             msgHeader.setPackageInfoField(this.parseIntFromBytes(data, 12, 4));
  92.             // byte[0-1] 消息包总数(word(16))
  93.             // tmp = new byte[2];
  94.             // System.arraycopy(data, 12, tmp, 0, 2);
  95.             // msgHeader.setTotalSubPackage(this.bitOperator.twoBytesToInteger(tmp));
  96.             msgHeader.setTotalSubPackage(this.parseIntFromBytes(data, 12, 2));
  97.  
  98.             // byte[2-3] 包序号(word(16)) 从 1 开始
  99.             // tmp = new byte[2];
  100.             // System.arraycopy(data, 14, tmp, 0, 2);
  101.             // msgHeader.setSubPackageSeq(this.bitOperator.twoBytesToInteger(tmp));
  102.             msgHeader.setSubPackageSeq(this.parseIntFromBytes(data, 12, 2));
  103.         }
  104.         return msgHeader;
  105.     }
  106.  
  107.     protected String parseStringFromBytes(byte[] data, int startIndex, int lenth) {
  108.         return this.parseStringFromBytes(data, startIndex, lenth, null);
  109.     }
  110.  
  111.     private String parseStringFromBytes(byte[] data, int startIndex, int lenth, String defaultVal) {
  112.         try {
  113.             byte[] tmp = new byte[lenth];
  114.             System.arraycopy(data, startIndex, tmp, 0, lenth);
  115.             return new String(tmp, "UTF-8");
  116.         } catch (Exception e) {
  117.             log.error("解析字符串出错:{}", e.getMessage());
  118.             e.printStackTrace();
  119.             return defaultVal;
  120.         }
  121.     }
  122.  
  123.     private String parseBcdStringFromBytes(byte[] data, int startIndex, int lenth) {
  124.         return this.parseBcdStringFromBytes(data, startIndex, lenth, null);
  125.     }
  126.  
  127.     private String parseBcdStringFromBytes(byte[] data, int startIndex, int lenth, String defaultVal) {
  128.         try {
  129.             byte[] tmp = new byte[lenth];
  130.             System.arraycopy(data, startIndex, tmp, 0, lenth);
  131.             return this.bcd8421Operater.bcd2String(tmp);
  132.         } catch (Exception e) {
  133.             log.error("解析BCD(8421码)出错:{}", e.getMessage());
  134.             e.printStackTrace();
  135.             return defaultVal;
  136.         }
  137.     }
  138.  
  139.     private int parseIntFromBytes(byte[] data, int startIndex, int length) {
  140.         return this.parseIntFromBytes(data, startIndex, length, 0);
  141.     }
  142.  
  143.     private int parseIntFromBytes(byte[] data, int startIndex, int length, int defaultVal) {
  144.         try {
  145.             // 字节数大于4,从起始索引开始向后处理4个字节,其余超出部分丢弃
  146.             final int len = length > 4 ? 4 : length;
  147.             byte[] tmp = new byte[len];
  148.             System.arraycopy(data, startIndex, tmp, 0, len);
  149.             return bitOperator.byteToInteger(tmp);
  150.         } catch (Exception e) {
  151.             log.error("解析整数出错:{}", e.getMessage());
  152.             e.printStackTrace();
  153.             return defaultVal;
  154.         }
  155.     }
  156. }
  •  

2.2.2 用到的工具类

2.2.2.1 BCD操作工具类

  1. package cn.hylexus.jt808.util;
  2.  
  3. public class BCD8421Operater {
  4.  
  5.     /**
  6.      * BCD字节数组===>String
  7.      * 
  8.      * @param bytes
  9.      * @return 十进制字符串
  10.      */
  11.     public String bcd2String(byte[] bytes) {
  12.         StringBuilder temp = new StringBuilder(bytes.length * 2);
  13.         for (int i = 0; i < bytes.length; i++) {
  14.             // 高四位
  15.             temp.append((bytes[i] & 0xf0) >>> 4);
  16.             // 低四位
  17.             temp.append(bytes[i] & 0x0f);
  18.         }
  19.         return temp.toString().substring(0, 1).equalsIgnoreCase("0") ? temp.toString().substring(1) : temp.toString();
  20.     }
  21.  
  22.     /**
  23.      * 字符串==>BCD字节数组
  24.      * 
  25.      * @param str
  26.      * @return BCD字节数组
  27.      */
  28.     public byte[] string2Bcd(String str) {
  29.         // 奇数,前补零
  30.         if ((str.length() & 0x1) == 1) {
  31.             str = "0" + str;
  32.         }
  33.  
  34.         byte ret[] = new byte[str.length() / 2];
  35.         byte bs[] = str.getBytes();
  36.         for (int i = 0; i < ret.length; i++) {
  37.  
  38.             byte high = ascII2Bcd(bs[2 * i]);
  39.             byte low = ascII2Bcd(bs[2 * i + 1]);
  40.  
  41.             // TODO 只遮罩BCD低四位?
  42.             ret[i] = (byte) ((high << 4) | low);
  43.         }
  44.         return ret;
  45.     }
  46.  
  47.     private byte ascII2Bcd(byte asc) {
  48.         if ((asc >= '0') && (asc <= '9'))
  49.             return (byte) (asc - '0');
  50.         else if ((asc >= 'A') && (asc <= 'F'))
  51.             return (byte) (asc - 'A' + 10);
  52.         else if ((asc >= 'a') && (asc <= 'f'))
  53.             return (byte) (asc - 'a' + 10);
  54.         else
  55.             return (byte) (asc - 48);
  56.     }
  57. }

2.2.2.2 位操作工具类

  1. package cn.hylexus.jt808.util;
  2.  
  3. import java.util.Arrays;
  4. import java.util.List;
  5.  
  6. public class BitOperator {
  7.  
  8.     /**
  9.      * 把一个整形该为byte
  10.      * 
  11.      * @param value
  12.      * @return
  13.      * @throws Exception
  14.      */
  15.     public byte integerTo1Byte(int value) {
  16.         return (byte) (value & 0xFF);
  17.     }
  18.  
  19.     /**
  20.      * 把一个整形该为1位的byte数组
  21.      * 
  22.      * @param value
  23.      * @return
  24.      * @throws Exception
  25.      */
  26.     public byte[] integerTo1Bytes(int value) {
  27.         byte[] result = new byte[1];
  28.         result[0] = (byte) (value & 0xFF);
  29.         return result;
  30.     }
  31.  
  32.     /**
  33.      * 把一个整形改为2位的byte数组
  34.      * 
  35.      * @param value
  36.      * @return
  37.      * @throws Exception
  38.      */
  39.     public byte[] integerTo2Bytes(int value) {
  40.         byte[] result = new byte[2];
  41.         result[0] = (byte) ((value >>> 8) & 0xFF);
  42.         result[1] = (byte) (value & 0xFF);
  43.         return result;
  44.     }
  45.  
  46.     /**
  47.      * 把一个整形改为3位的byte数组
  48.      * 
  49.      * @param value
  50.      * @return
  51.      * @throws Exception
  52.      */
  53.     public byte[] integerTo3Bytes(int value) {
  54.         byte[] result = new byte[3];
  55.         result[0] = (byte) ((value >>> 16) & 0xFF);
  56.         result[1] = (byte) ((value >>> 8) & 0xFF);
  57.         result[2] = (byte) (value & 0xFF);
  58.         return result;
  59.     }
  60.  
  61.     /**
  62.      * 把一个整形改为4位的byte数组
  63.      * 
  64.      * @param value
  65.      * @return
  66.      * @throws Exception
  67.      */
  68.     public byte[] integerTo4Bytes(int value){
  69.         byte[] result = new byte[4];
  70.         result[0] = (byte) ((value >>> 24) & 0xFF);
  71.         result[1] = (byte) ((value >>> 16) & 0xFF);
  72.         result[2] = (byte) ((value >>> 8) & 0xFF);
  73.         result[3] = (byte) (value & 0xFF);
  74.         return result;
  75.     }
  76.  
  77.     /**
  78.      * 把byte[]转化位整形,通常为指令用
  79.      * 
  80.      * @param value
  81.      * @return
  82.      * @throws Exception
  83.      */
  84.     public int byteToInteger(byte[] value) {
  85.         int result;
  86.         if (value.length == 1) {
  87.             result = oneByteToInteger(value[0]);
  88.         } else if (value.length == 2) {
  89.             result = twoBytesToInteger(value);
  90.         } else if (value.length == 3) {
  91.             result = threeBytesToInteger(value);
  92.         } else if (value.length == 4) {
  93.             result = fourBytesToInteger(value);
  94.         } else {
  95.             result = fourBytesToInteger(value);
  96.         }
  97.         return result;
  98.     }
  99.  
  100.     /**
  101.      * 把一个byte转化位整形,通常为指令用
  102.      * 
  103.      * @param value
  104.      * @return
  105.      * @throws Exception
  106.      */
  107.     public int oneByteToInteger(byte value) {
  108.         return (int) value & 0xFF;
  109.     }
  110.  
  111.     /**
  112.      * 把一个2位的数组转化位整形
  113.      * 
  114.      * @param value
  115.      * @return
  116.      * @throws Exception
  117.      */
  118.     public int twoBytesToInteger(byte[] value) {
  119.         // if (value.length < 2) {
  120.         // throw new Exception("Byte array too short!");
  121.         // }
  122.         int temp0 = value[0] & 0xFF;
  123.         int temp1 = value[1] & 0xFF;
  124.         return ((temp0 << 8) + temp1);
  125.     }
  126.  
  127.     /**
  128.      * 把一个3位的数组转化位整形
  129.      * 
  130.      * @param value
  131.      * @return
  132.      * @throws Exception
  133.      */
  134.     public int threeBytesToInteger(byte[] value) {
  135.         int temp0 = value[0] & 0xFF;
  136.         int temp1 = value[1] & 0xFF;
  137.         int temp2 = value[2] & 0xFF;
  138.         return ((temp0 << 16) + (temp1 << 8) + temp2);
  139.     }
  140.  
  141.     /**
  142.      * 把一个4位的数组转化位整形,通常为指令用
  143.      * 
  144.      * @param value
  145.      * @return
  146.      * @throws Exception
  147.      */
  148.     public int fourBytesToInteger(byte[] value) {
  149.         // if (value.length < 4) {
  150.         // throw new Exception("Byte array too short!");
  151.         // }
  152.         int temp0 = value[0] & 0xFF;
  153.         int temp1 = value[1] & 0xFF;
  154.         int temp2 = value[2] & 0xFF;
  155.         int temp3 = value[3] & 0xFF;
  156.         return ((temp0 << 24) + (temp1 << 16) + (temp2 << 8) + temp3);
  157.     }
  158.  
  159.     /**
  160.      * 把一个4位的数组转化位整形
  161.      * 
  162.      * @param value
  163.      * @return
  164.      * @throws Exception
  165.      */
  166.     public long fourBytesToLong(byte[] value) throws Exception {
  167.         // if (value.length < 4) {
  168.         // throw new Exception("Byte array too short!");
  169.         // }
  170.         int temp0 = value[0] & 0xFF;
  171.         int temp1 = value[1] & 0xFF;
  172.         int temp2 = value[2] & 0xFF;
  173.         int temp3 = value[3] & 0xFF;
  174.         return (((long) temp0 << 24) + (temp1 << 16) + (temp2 << 8) + temp3);
  175.     }
  176.  
  177.     /**
  178.      * 把一个数组转化长整形
  179.      * 
  180.      * @param value
  181.      * @return
  182.      * @throws Exception
  183.      */
  184.     public long bytes2Long(byte[] value) {
  185.         long result = 0;
  186.         int len = value.length;
  187.         int temp;
  188.         for (int i = 0; i < len; i++) {
  189.             temp = (len - 1 - i) * 8;
  190.             if (temp == 0) {
  191.                 result += (value[i] & 0x0ff);
  192.             } else {
  193.                 result += (value[i] & 0x0ff) << temp;
  194.             }
  195.         }
  196.         return result;
  197.     }
  198.  
  199.     /**
  200.      * 把一个长整形改为byte数组
  201.      * 
  202.      * @param value
  203.      * @return
  204.      * @throws Exception
  205.      */
  206.     public byte[] longToBytes(long value){
  207.         return longToBytes(value, 8);
  208.     }
  209.  
  210.     /**
  211.      * 把一个长整形改为byte数组
  212.      * 
  213.      * @param value
  214.      * @return
  215.      * @throws Exception
  216.      */
  217.     public byte[] longToBytes(long value, int len) {
  218.         byte[] result = new byte[len];
  219.         int temp;
  220.         for (int i = 0; i < len; i++) {
  221.             temp = (len - 1 - i) * 8;
  222.             if (temp == 0) {
  223.                 result[i] += (value & 0x0ff);
  224.             } else {
  225.                 result[i] += (value >>> temp) & 0x0ff;
  226.             }
  227.         }
  228.         return result;
  229.     }
  230.  
  231.     /**
  232.      * 得到一个消息ID
  233.      * 
  234.      * @return
  235.      * @throws Exception
  236.      */
  237.     public byte[] generateTransactionID() throws Exception {
  238.         byte[] id = new byte[16];
  239.         System.arraycopy(integerTo2Bytes((int) (Math.random() * 65536)), 0, id, 0, 2);
  240.         System.arraycopy(integerTo2Bytes((int) (Math.random() * 65536)), 0, id, 2, 2);
  241.         System.arraycopy(integerTo2Bytes((int) (Math.random() * 65536)), 0, id, 4, 2);
  242.         System.arraycopy(integerTo2Bytes((int) (Math.random() * 65536)), 0, id, 6, 2);
  243.         System.arraycopy(integerTo2Bytes((int) (Math.random() * 65536)), 0, id, 8, 2);
  244.         System.arraycopy(integerTo2Bytes((int) (Math.random() * 65536)), 0, id, 10, 2);
  245.         System.arraycopy(integerTo2Bytes((int) (Math.random() * 65536)), 0, id, 12, 2);
  246.         System.arraycopy(integerTo2Bytes((int) (Math.random() * 65536)), 0, id, 14, 2);
  247.         return id;
  248.     }
  249.  
  250.     /**
  251.      * 把IP拆分位int数组
  252.      * 
  253.      * @param ip
  254.      * @return
  255.      * @throws Exception
  256.      */
  257.     public int[] getIntIPValue(String ip) throws Exception {
  258.         String[] sip = ip.split("[.]");
  259.         // if (sip.length != 4) {
  260.         // throw new Exception("error IPAddress");
  261.         // }
  262.         int[] intIP = { Integer.parseInt(sip[0]), Integer.parseInt(sip[1]), Integer.parseInt(sip[2]),
  263.                 Integer.parseInt(sip[3]) };
  264.         return intIP;
  265.     }
  266.  
  267.     /**
  268.      * 把byte类型IP地址转化位字符串
  269.      * 
  270.      * @param address
  271.      * @return
  272.      * @throws Exception
  273.      */
  274.     public String getStringIPValue(byte[] address) throws Exception {
  275.         int first = this.oneByteToInteger(address[0]);
  276.         int second = this.oneByteToInteger(address[1]);
  277.         int third = this.oneByteToInteger(address[2]);
  278.         int fourth = this.oneByteToInteger(address[3]);
  279.  
  280.         return first + "." + second + "." + third + "." + fourth;
  281.     }
  282.  
  283.     /**
  284.      * 合并字节数组
  285.      * 
  286.      * @param first
  287.      * @param rest
  288.      * @return
  289.      */
  290.     public byte[] concatAll(byte[] first, byte[]... rest) {
  291.         int totalLength = first.length;
  292.         for (byte[] array : rest) {
  293.             if (array != null) {
  294.                 totalLength += array.length;
  295.             }
  296.         }
  297.         byte[] result = Arrays.copyOf(first, totalLength);
  298.         int offset = first.length;
  299.         for (byte[] array : rest) {
  300.             if (array != null) {
  301.                 System.arraycopy(array, 0, result, offset, array.length);
  302.                 offset += array.length;
  303.             }
  304.         }
  305.         return result;
  306.     }
  307.  
  308.     /**
  309.      * 合并字节数组
  310.      * 
  311.      * @param rest
  312.      * @return
  313.      */
  314.     public byte[] concatAll(List<byte[]> rest) {
  315.         int totalLength = 0;
  316.         for (byte[] array : rest) {
  317.             if (array != null) {
  318.                 totalLength += array.length;
  319.             }
  320.         }
  321.         byte[] result = new byte[totalLength];
  322.         int offset = 0;
  323.         for (byte[] array : rest) {
  324.             if (array != null) {
  325.                 System.arraycopy(array, 0, result, offset, array.length);
  326.                 offset += array.length;
  327.             }
  328.         }
  329.         return result;
  330.     }
  331.  
  332.     public float byte2Float(byte[] bs) {
  333.         return Float.intBitsToFloat(
  334.                 (((bs[3] & 0xFF) << 24) + ((bs[2] & 0xFF) << 16) + ((bs[1] & 0xFF) << 8) + (bs[0] & 0xFF)));
  335.     }
  336.  
  337.     public float byteBE2Float(byte[] bytes) {
  338.         int l;
  339.         l = bytes[0];
  340.         l &= 0xff;
  341.         l |= ((long) bytes[1] << 8);
  342.         l &= 0xffff;
  343.         l |= ((long) bytes[2] << 16);
  344.         l &= 0xffffff;
  345.         l |= ((long) bytes[3] << 24);
  346.         return Float.intBitsToFloat(l);
  347.     }
  348.  
  349.     public int getCheckSum4JT808(byte[] bs, int start, int end) {
  350.         if (start < 0 || end > bs.length)
  351.             throw new ArrayIndexOutOfBoundsException("getCheckSum4JT808 error : index out of bounds(start=" + start
  352.                     + ",end=" + end + ",bytes length=" + bs.length + ")");
  353.         int cs = 0;
  354.         for (int i = start; i < end; i++) {
  355.             cs ^= bs[i];
  356.         }
  357.         return cs;
  358.     }
  359.  
  360.     public int getBitRange(int number, int start, int end) {
  361.         if (start < 0)
  362.             throw new IndexOutOfBoundsException("min index is 0,but start = " + start);
  363.         if (end >= Integer.SIZE)
  364.             throw new IndexOutOfBoundsException("max index is " + (Integer.SIZE - 1) + ",but end = " + end);
  365.  
  366.         return (number << Integer.SIZE - (end + 1)) >>> Integer.SIZE - (end - start + 1);
  367.     }
  368.  
  369.     public int getBitAt(int number, int index) {
  370.         if (index < 0)
  371.             throw new IndexOutOfBoundsException("min index is 0,but " + index);
  372.         if (index >= Integer.SIZE)
  373.             throw new IndexOutOfBoundsException("max index is " + (Integer.SIZE - 1) + ",but " + index);
  374.  
  375.         return ((1 << index) & number) >> index;
  376.     }
  377.  
  378.     public int getBitAtS(int number, int index) {
  379.         String s = Integer.toBinaryString(number);
  380.         return Integer.parseInt(s.charAt(index) + "");
  381.     }
  382.  
  383.     @Deprecated
  384.     public int getBitRangeS(int number, int start, int end) {
  385.         String s = Integer.toBinaryString(number);
  386.         StringBuilder sb = new StringBuilder(s);
  387.         while (sb.length() < Integer.SIZE) {
  388.             sb.insert(0, "0");
  389.         }
  390.         String tmp = sb.reverse().substring(start, end + 1);
  391.         sb = new StringBuilder(tmp);
  392.         return Integer.parseInt(sb.reverse().toString(), 2);
  393.     }
  394. }

2.3 和netty结合

2.3.1 netty处理器链

  1. import java.util.concurrent.TimeUnit;
  2.  
  3. import org.slf4j.Logger;
  4. import org.slf4j.LoggerFactory;
  5.  
  6. import cn.kkbc.tpms.tcp.service.TCPServerHandler;
  7. import io.netty.bootstrap.ServerBootstrap;
  8. import io.netty.buffer.Unpooled;
  9. import io.netty.channel.ChannelFuture;
  10. import io.netty.channel.ChannelInitializer;
  11. import io.netty.channel.ChannelOption;
  12. import io.netty.channel.EventLoopGroup;
  13. import io.netty.channel.nio.NioEventLoopGroup;
  14. import io.netty.channel.socket.SocketChannel;
  15. import io.netty.channel.socket.nio.NioServerSocketChannel;
  16. import io.netty.handler.codec.DelimiterBasedFrameDecoder;
  17. import io.netty.handler.timeout.IdleStateHandler;
  18. import io.netty.util.concurrent.Future;
  19.  
  20. public class TCPServer2 {
  21.  
  22.     private Logger log = LoggerFactory.getLogger(getClass());
  23.     private volatile boolean isRunning = false;
  24.  
  25.     private EventLoopGroup bossGroup = null;
  26.     private EventLoopGroup workerGroup = null;
  27.     private int port;
  28.  
  29.     public TCPServer2() {
  30.     }
  31.  
  32.     public TCPServer2(int port) {
  33.         this();
  34.         this.port = port;
  35.     }
  36.  
  37.     private void bind() throws Exception {
  38.         this.bossGroup = new NioEventLoopGroup();
  39.         this.workerGroup = new NioEventLoopGroup();
  40.         ServerBootstrap serverBootstrap = new ServerBootstrap();
  41.         serverBootstrap.group(bossGroup, workerGroup)//
  42.                 .channel(NioServerSocketChannel.class) //
  43.                 .childHandler(new ChannelInitializer<SocketChannel>() { //
  44.                     @Override
  45.                     public void initChannel(SocketChannel ch) throws Exception {
  46.                         //超过15分钟未收到客户端消息则自动断开客户端连接
  47.                         ch.pipeline().addLast("idleStateHandler",
  48.                                 new IdleStateHandler(15, 0, 0, TimeUnit.MINUTES));
  49.                         //ch.pipeline().addLast(new Decoder4LoggingOnly());
  50.                         // 1024表示单条消息的最大长度，解码器在查找分隔符的时候，达到该长度还没找到的话会抛异常
  51.                         ch.pipeline().addLast(
  52.                                 new DelimiterBasedFrameDecoder(1024, Unpooled.copiedBuffer(new byte[] { 0x7e }),
  53.                                         Unpooled.copiedBuffer(new byte[] { 0x7e, 0x7e })));
  54.                         ch.pipeline().addLast(new TCPServerHandler());
  55.                     }
  56.                 }).option(ChannelOption.SO_BACKLOG, 128) //
  57.                 .childOption(ChannelOption.SO_KEEPALIVE, true);
  58.  
  59.         this.log.info("TCP服务启动完毕,port={}", this.port);
  60.         ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
  61.  
  62.         channelFuture.channel().closeFuture().sync();
  63.     }
  64.  
  65.     public synchronized void startServer() {
  66.         if (this.isRunning) {
  67.             throw new IllegalStateException(this.getName() + " is already started .");
  68.         }
  69.         this.isRunning = true;
  70.  
  71.         new Thread(() -> {
  72.             try {
  73.                 this.bind();
  74.             } catch (Exception e) {
  75.                 this.log.info("TCP服务启动出错:{}", e.getMessage());
  76.                 e.printStackTrace();
  77.             }
  78.         }, this.getName()).start();
  79.     }
  80.  
  81.     public synchronized void stopServer() {
  82.         if (!this.isRunning) {
  83.             throw new IllegalStateException(this.getName() + " is not yet started .");
  84.         }
  85.         this.isRunning = false;
  86.  
  87.         try {
  88.             Future<?> future = this.workerGroup.shutdownGracefully().await();
  89.             if (!future.isSuccess()) {
  90.                 log.error("workerGroup 无法正常停止:{}", future.cause());
  91.             }
  92.  
  93.             future = this.bossGroup.shutdownGracefully().await();
  94.             if (!future.isSuccess()) {
  95.                 log.error("bossGroup 无法正常停止:{}", future.cause());
  96.             }
  97.         } catch (InterruptedException e) {
  98.             e.printStackTrace();
  99.         }
  100.  
  101.         this.log.info("TCP服务已经停止...");
  102.     }
  103.  
  104.     private String getName() {
  105.         return "TCP-Server";
  106.     }
  107.  
  108.     public static void main(String[] args) throws Exception {
  109.         TCPServer2 server = new TCPServer2(20048);
  110.         server.startServer();
  111.  
  112.         // Thread.sleep(3000);
  113.         // server.stopServer();
  114.     }
  115.  
  116. }
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102
  • 103
  • 104
  • 105
  • 106
  • 107
  • 108
  • 109
  • 110
  • 111
  • 112
  • 113
  • 114
  • 115
  • 116

2.3.2 netty针对于JT808的消息处理器

  1. package cn.hylexus.jt808.service;
  2.  
  3. import org.slf4j.Logger;
  4. import org.slf4j.LoggerFactory;
  5.  
  6. import cn.hylexus.jt808.server.SessionManager;
  7. import cn.hylexus.jt808.service.codec.MsgDecoder;
  8. import cn.hylexus.jt808.vo.PackageData;
  9. import cn.hylexus.jt808.vo.Session;
  10. import io.netty.buffer.ByteBuf;
  11. import io.netty.channel.ChannelHandlerContext;
  12. import io.netty.channel.ChannelInboundHandlerAdapter;
  13. import io.netty.handler.timeout.IdleState;
  14. import io.netty.handler.timeout.IdleStateEvent;
  15. import io.netty.util.ReferenceCountUtil;
  16.  
  17. public class TCPServerHandler extends ChannelInboundHandlerAdapter { // (1)
  18.  
  19.     private final Logger logger = LoggerFactory.getLogger(getClass());
  20.  
  21.     // 一个维护客户端连接的类
  22.     private final SessionManager sessionManager;
  23.     private MsgDecoder decoder = new MsgDecoder();
  24.  
  25.     public TCPServerHandler() {
  26.         this.sessionManager = SessionManager.getInstance();
  27.     }
  28.  
  29.     @Override
  30.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws InterruptedException { // (2)
  31.         try {
  32.             ByteBuf buf = (ByteBuf) msg;
  33.             if (buf.readableBytes() <= 0) {
  34.                 // ReferenceCountUtil.safeRelease(msg);
  35.                 return;
  36.             }
  37.  
  38.             byte[] bs = new byte[buf.readableBytes()];
  39.             buf.readBytes(bs);
  40.  
  41.             PackageData jt808Msg = this.decoder.queueElement2PackageData(bs);
  42.             // 处理客户端消息
  43.             this.processClientMsg(jt808Msg);
  44.         } finally {
  45.             release(msg);
  46.         }
  47.     }
  48.  
  49.     private void processClientMsg(PackageData jt808Msg) {
  50.         // TODO 更加消息ID的不同,分别实现自己的业务逻辑
  51.         if (jt808Msg.getMsgHeader().getMsgId() == 0x900) {
  52.             // TODO ...
  53.         } else if (jt808Msg.getMsgHeader().getMsgId() == 0x9001) {
  54.             // TODO ...
  55.         }
  56.         // else if(){}
  57.         // else if(){}
  58.         // else if(){}
  59.         // else if(){}
  60.         // ...
  61.         else {
  62.             logger.error("位置消息,消息ID={}", jt808Msg.getMsgHeader().getMsgId());
  63.         }
  64.     }
  65.  
  66.     @Override
  67.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // (4)
  68.         logger.error("发生异常:{}", cause.getMessage());
  69.         cause.printStackTrace();
  70.     }
  71.  
  72.     @Override
  73.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  74.         Session session = Session.buildSession(ctx.channel());
  75.         sessionManager.put(session.getId(), session);
  76.         logger.debug("终端连接:{}", session);
  77.     }
  78.  
  79.     @Override
  80.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  81.         final String sessionId = ctx.channel().id().asLongText();
  82.         Session session = sessionManager.findBySessionId(sessionId);
  83.         this.sessionManager.removeBySessionId(sessionId);
  84.         logger.debug("终端断开连接:{}", session);
  85.         ctx.channel().close();
  86.         // ctx.close();
  87.     }
  88.  
  89.     @Override
  90.     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
  91.         if (IdleStateEvent.class.isAssignableFrom(evt.getClass())) {
  92.             IdleStateEvent event = (IdleStateEvent) evt;
  93.             if (event.state() == IdleState.READER_IDLE) {
  94.                 Session session = this.sessionManager.removeBySessionId(Session.buildId(ctx.channel()));
  95.                 logger.error("服务器主动断开连接:{}", session);
  96.                 ctx.close();
  97.             }
  98.         }
  99.     }
  100.  
  101.     private void release(Object msg) {
  102.         try {
  103.             ReferenceCountUtil.release(msg);
  104.         } catch (Exception e) {
  105.             e.printStackTrace();
  106.         }
  107.     }
  108. }

2.3.3 用到的其他类

  1. package cn.hylexus.jt808.server;
  2.  
  3. import java.util.List;
  4. import java.util.Map;
  5. import java.util.Map.Entry;
  6. import java.util.Set;
  7. import java.util.concurrent.ConcurrentHashMap;
  8. import java.util.function.BiConsumer;
  9. import java.util.stream.Collectors;
  10.  
  11. import cn.hylexus.jt808.vo.Session;
  12.  
  13. public class SessionManager {
  14.  
  15.     private static volatile SessionManager instance = null;
  16.     // netty生成的sessionID和Session的对应关系
  17.     private Map<String, Session> sessionIdMap;
  18.     // 终端手机号和netty生成的sessionID的对应关系
  19.     private Map<String, String> phoneMap;
  20.  
  21.     public static SessionManager getInstance() {
  22.         if (instance == null) {
  23.             synchronized (SessionManager.class) {
  24.                 if (instance == null) {
  25.                     instance = new SessionManager();
  26.                 }
  27.             }
  28.         }
  29.         return instance;
  30.     }
  31.  
  32.     public SessionManager() {
  33.         this.sessionIdMap = new ConcurrentHashMap<>();
  34.         this.phoneMap = new ConcurrentHashMap<>();
  35.     }
  36.  
  37.     public boolean containsKey(String sessionId) {
  38.         return sessionIdMap.containsKey(sessionId);
  39.     }
  40.  
  41.     public boolean containsSession(Session session) {
  42.         return sessionIdMap.containsValue(session);
  43.     }
  44.  
  45.     public Session findBySessionId(String id) {
  46.         return sessionIdMap.get(id);
  47.     }
  48.  
  49.     public Session findByTerminalPhone(String phone) {
  50.         String sessionId = this.phoneMap.get(phone);
  51.         if (sessionId == null)
  52.             return null;
  53.         return this.findBySessionId(sessionId);
  54.     }
  55.  
  56.     public synchronized Session put(String key, Session value) {
  57.         if (value.getTerminalPhone() != null && !"".equals(value.getTerminalPhone().trim())) {
  58.             this.phoneMap.put(value.getTerminalPhone(), value.getId());
  59.         }
  60.         return sessionIdMap.put(key, value);
  61.     }
  62.  
  63.     public synchronized Session removeBySessionId(String sessionId) {
  64.         if (sessionId == null)
  65.             return null;
  66.         Session session = sessionIdMap.remove(sessionId);
  67.         if (session == null)
  68.             return null;
  69.         if (session.getTerminalPhone() != null)
  70.             this.phoneMap.remove(session.getTerminalPhone());
  71.         return session;
  72.     }
  73.  
  74.     public Set<String> keySet() {
  75.         return sessionIdMap.keySet();
  76.     }
  77.  
  78.     public void forEach(BiConsumer<? super String, ? super Session> action) {
  79.         sessionIdMap.forEach(action);
  80.     }
  81.  
  82.     public Set<Entry<String, Session>> entrySet() {
  83.         return sessionIdMap.entrySet();
  84.     }
  85.  
  86.     public List<Session> toList() {
  87.         return this.sessionIdMap.entrySet().stream().map(e -> e.getValue()).collect(Collectors.toList());
  88.     }
  89.  
  90. }

3 demo级别java示例

请移步: https://github.com/hylexus/jt-808-protocol

相关技术文章

最新源码

下载排行榜
点击QQ咨询
开通会员
返回顶部
×
微信扫码支付 元
微信扫码支付
确定支付下载
请使用微信描二维码支付
×

提示信息

×

选择支付方式

  • 微信支付
  • 支付宝付款
确定支付下载