原文:

https://blog.csdn.net/fox_mt/article/details/52806560?utm_medium=distribute.pc_relevant.none-task-blog-searchFromBaidu-5.control&depth_1-utm_source=distribute.pc_relevant.none-task-blog-searchFromBaidu-5.control

最近使用JT809協議進行資料對接，遇到了不少問題，度娘谷歌都不好使，找不到很好的客戶端實現程式碼的例子，只能苦逼的自己悶頭弄，現在特意寫篇帖子，希望能幫助一些人

說正經的：

背景：跟某公司做資料對接，將本公司的一些資訊推送到接收端

要求：建立tcp連結，使用接收端提供的使用者名稱密碼等資訊 先登入，登入成功後推送資料，資料採用JT809標準協議

實現語言：java

下面介紹具體實現，包涵完整程式碼

在這之前，最好先下載jt809協議，研究研究，網上就有，我找到的裡面還有一些錯誤別，一度讓我覺得是個盜版貨

首先說下整體結構：一個tcp客戶端，一個Decoder，一個心跳Handler，一個處理反饋的handler，一個訊息物件

通過tcp客戶端連結制定ip地址和埠，進行登陸，心跳handler負責在傳送資料空閒時向目標伺服器傳送請求保持包；decoder負責解碼組裝物件，反饋handler負責處理收到的反饋資訊，訊息物件我就不多說了

下面看程式碼：

TCPclient：

由於時間緊迫，程式碼結構有些臃腫，我使用的tcpclient為兩個，一個是基類，一個子類，其實這兩個可以合併到一起，這裡有心的朋友自己去和，下面來看下tcpclient基類的程式碼,其中有一些模組的引用因為涉及到公司資訊，所以我刪除掉了，大家參考的時候，引用自己編寫的就可以了，

1. //包路徑，自己填
2. import java.net.InetSocketAddress;
3. import java.util.concurrent.Executor;
4. import java.util.concurrent.Executors;
6. import org.jboss.netty.bootstrap.ClientBootstrap;
7. import org.jboss.netty.buffer.ChannelBuffer;
8. import org.jboss.netty.buffer.ChannelBuffers;
9. import org.jboss.netty.channel.Channel;
10. import org.jboss.netty.channel.ChannelFuture;
11. import org.jboss.netty.channel.ChannelPipeline;
12. import org.jboss.netty.channel.ChannelPipelineFactory;
13. import org.jboss.netty.channel.Channels;
14. import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
15. import org.jboss.netty.handler.logging.LoggingHandler;
16. import org.jboss.netty.handler.timeout.IdleStateHandler;
17. import org.jboss.netty.logging.InternalLogLevel;
18. import org.jboss.netty.util.HashedWheelTimer;
19. import org.slf4j.Logger;
20. import org.slf4j.LoggerFactory;
24. public class TcpClient{
26. private static final Logger LOG = LoggerFactory.getLogger(TcpClient.class);
28. private static final int DEFAULT_PORT = 9000;
30. private long connectTimeoutMillis = 3000;
32. private int port = DEFAULT_PORT;
34. private boolean tcpNoDelay = false;
36. private boolean reuseAddress = true;
38. private boolean keepAlive = true;
40. private int workerCount = 4;
42. private ClientBootstrap bootstrap = null;
44. private static Channel channel = null;
46. private Executor bossExecutor = Executors.newCachedThreadPool();
48. private Executor workerExecutor = Executors.newCachedThreadPool();
51. private static TcpClient instance = new TcpClient();
53. private TcpClient() {
54. init();
55. }
57. public static TcpClient getInstence(){
58. return instance;
59. }
62. public void init() {
64. bootstrap = new ClientBootstrap(new NioClientSocketChannelFactory(
65. bossExecutor, workerExecutor, workerCount));
66. bootstrap.setOption("tcpNoDelay", tcpNoDelay);
67. bootstrap.setOption("connectTimeoutMillis", connectTimeoutMillis);
68. bootstrap.setOption("reuseAddress", reuseAddress);
69. bootstrap.setOption("keepAlive", keepAlive);
70. }
73. public Channel getChannel(String address, int port) {
75. if (null == channel || !channel.isOpen()) {
76. bootstrap.setOption("writeBufferHighWaterMark", 64 * 1024);
77. bootstrap.setOption("writeBufferLowWaterMark", 32 * 1024);
78. bootstrap.setPipelineFactory(new ChannelPipelineFactory(){
79. @Override
80. public ChannelPipeline getPipeline() throws Exception {
81. ChannelPipeline pipeline = Channels.pipeline();
82. // pipeline.addLast("loging", new LoggingHandler(InternalLogLevel.ERROR)); 列印日誌資訊，上線穩定後可去掉
83. pipeline.addLast("timeout", new IdleStateHandler(new HashedWheelTimer(), 10, 60, 0));//設定空閒心跳機制
84. pipeline.addLast("heartbeat", new HeartBeatHandler());//心跳傳送包處理handler
85. pipeline.addLast("decode", new Decoder());//解碼
86. pipeline.addLast("loginHandler", new RecevieHandler());//反饋資料處理
87. return pipeline;
88. }
89. });
90. ChannelFuture future = bootstrap.connect(new InetSocketAddress(
91. address, port));
92. future.awaitUninterruptibly();
93. if (future.isSuccess()) {
94. channel = future.getChannel();
95. } else {
96. throw new FrameworkRuntimeException(future.getCause());
97. }
98. }
100. return channel;
101. }
105. public long getConnectTimeoutMillis() {
106. return connectTimeoutMillis;
107. }
109. public void setConnectTimeoutMillis(long connectTimeoutMillis) {
110. this.connectTimeoutMillis = connectTimeoutMillis;
111. }
113. public int getPort() {
114. return port;
115. }
117. public void setPort(int port) {
118. this.port = port;
119. }
121. public boolean isTcpNoDelay() {
122. return tcpNoDelay;
123. }
125. public void setTcpNoDelay(boolean tcpNoDelay) {
126. this.tcpNoDelay = tcpNoDelay;
127. }
129. public boolean isReuseAddress() {
130. return reuseAddress;
131. }
133. public void setReuseAddress(boolean reuseAddress) {
134. this.reuseAddress = reuseAddress;
135. }
137. public boolean isKeepAlive() {
138. return keepAlive;
139. }
141. public void setKeepAlive(boolean keepAlive) {
142. this.keepAlive = keepAlive;
143. }
145. public int getWorkerCount() {
146. return workerCount;
147. }
149. public void setWorkerCount(int workerCount) {
150. this.workerCount = workerCount;
151. }
153. }

上面的配置，用過netty都不會陌生，就不多說了，下面看下子類的實現

3. import java.nio.charset.Charset;
4. import java.text.NumberFormat;
5. import java.util.Calendar;
6. import java.util.Date;
7. import java.util.Properties;
9. import org.apache.commons.lang3.StringUtils;
10. import org.jboss.netty.buffer.ChannelBuffer;
11. import org.jboss.netty.buffer.ChannelBuffers;
12. import org.jboss.netty.channel.Channel;
13. import org.slf4j.Logger;
14. import org.slf4j.LoggerFactory;
17. public class TcpClientDemo {
19. private static Logger LOGGER = LoggerFactory.getLogger(TcpClientDemo.class);
21. /**
22. * 交委指定本公司接入碼
23. */
24. public static int PLANT_CODE;
25. /**
26. * 交委指定本公司使用者名稱
27. */
28. public static int COM_ID;
29. /**
30. * 交委指定本公司密碼
31. */
32. public static String COM_PWD;
33. public static String LONGINSTATUS = "";
34. public static String LOGINING = "logining";
35. private static int LOGIN_FLAG = 0;
36. private static String DOWN_LINK_IP = "127.0.0.1";
38. //初始化基類
39. private static TcpClient tcpClient = TcpClient.getInstence();
40. //初始化
41. private static TcpClientDemo tcpClientDemo = new TcpClientDemo();
42. //初始化配置,將密碼相關的內容寫在裡面
43. private static Properties property = PropertiesReader.getProperties("properName", true);
44. //初始化channel,以便心跳機制及時登入
45. private Channel channel = tcpClient.getChannel(Constants.TCP_ADDRESS, Constants.TCP_PORT);
48. public static TcpClientDemo getInstance(){
49. //加在車輛資料
50. InitVehicleData.initVehicleMap();
51. //獲取本機IP對應的使用者名稱密碼,IpUtils自己實現一個，就是獲取本地IP的，因為有的城市的交委會給每個伺服器一個賬號密碼
52. String localIp = IpUtils.getLocalIp();
53. if(StringUtils.isNotBlank(localIp)){
54. String properties = property.getProperty(localIp);
55. if(StringUtils.isNotBlank(localIp)){
56. String[] pros = properties.split(":");
57. PLANT_CODE = Integer.parseInt(pros[0]);
58. COM_ID = Integer.parseInt(pros[1]);
59. COM_PWD = pros[2];
60. }
61. }else{
62. LOGGER.error("獲取本機IP異常");
63. }
64. return tcpClientDemo;
65. }
67. /**
68. * 判斷是否登入
69. * boolean
70. * @return
71. * 2016年10月12日 by fox_mt
72. */
73. public boolean isLogined(){
74. return Constants.LOGIN_SUCCESS.equals(LONGINSTATUS);
75. }
77. /**
78. * 登入交委接入平臺
79. * boolean
80. * @return
81. * 2016年9月28日 by fox_mt
82. */
83. public boolean login2Gov(){
85. boolean success = false;
86. if(!Constants.LOGIN_SUCCESS.equals(LONGINSTATUS) && !LOGINING.equals(LONGINSTATUS)){
87. //開始登入
88. Message msg = new Message(JT809Constants.UP_CONNECT_REQ);
89. ChannelBuffer buffer = ChannelBuffers.buffer(46);
90. buffer.writeInt(COM_ID);//4
92. byte[] pwd = getBytesWithLengthAfter(8, COM_PWD.getBytes());
93. buffer.writeBytes(pwd);//8
95. byte[] ip = getBytesWithLengthAfter(32, DOWN_LINK_IP.getBytes());
96. buffer.writeBytes(ip);//32
98. buffer.writeShort((short)Constants.TCP_RESULT_PORT);//2
99. msg.setMsgBody(buffer);
100. channel = tcpClient.getChannel(Constants.TCP_ADDRESS_FUJIAN, Constants.TCP_PORT_FUJIAN);
101. channel.write(buildMessage(msg));
102. LONGINSTATUS = LOGINING;
103. }
104. return success;
105. }
107. public static ChannelBuffer buildMessage(Message msg){
108. int bodyLength = 0 ;
109. if(null != msg.getMsgBody()){
110. bodyLength = msg.getMsgBody().readableBytes();
111. }
112. msg.setMsgGesscenterId(PLANT_CODE);
113. ChannelBuffer buffer = ChannelBuffers.dynamicBuffer(bodyLength + Message.MSG_FIX_LENGTH);
115. ChannelBuffer headBuffer = ChannelBuffers.buffer(22);
116. //---資料頭
117. headBuffer.writeInt(buffer.capacity() - 1);//4
118. headBuffer.writeInt(msg.getMsgSn());//4
119. headBuffer.writeShort((short)msg.getMsgId());//2
120. headBuffer.writeInt(msg.getMsgGesscenterId());//4
121. headBuffer.writeBytes(msg.getVersionFlag());//3
122. headBuffer.writeByte(0);//1
123. headBuffer.writeInt(10);//4
124. buffer.writeBytes(headBuffer);
125. //---資料體
126. if(null != msg.getMsgBody()){
127. buffer.writeBytes(msg.getMsgBody());
128. }
129. ChannelBuffer finalBuffer = ChannelBuffers.copiedBuffer(buffer);
130. //--crc校驗碼
131. byte[] b = ChannelBuffers.buffer(finalBuffer.readableBytes()).array();
132. finalBuffer.getBytes(0, b);
134. int crcValue = CRC16CCITT.crc16(b);
135. finalBuffer.writeShort((short)crcValue);//2
136. //轉義
137. byte[] bytes = ChannelBuffers.copiedBuffer(finalBuffer).array();
138. ChannelBuffer headFormatedBuffer = ChannelBuffers.dynamicBuffer(finalBuffer.readableBytes());
139. formatBuffer(bytes, headFormatedBuffer);
140. ChannelBuffer buffera = ChannelBuffers.buffer(headFormatedBuffer.readableBytes() + 2);
141. buffera.writeByte(Message.MSG_HEAD);
142. buffera.writeBytes(headFormatedBuffer);
143. buffera.writeByte(Message.MSG_TALL);
144. return ChannelBuffers.copiedBuffer(buffera);
145. }
148. /**
149. * 傳送資料到交委接入平臺
150. * boolean
151. * @param awsVo
152. * @return
153. * 2016年9月28日 by fox_mt
154. */
155. public boolean sendMsg2Gov(Idc2AwsGpsVo awsVo){
156. boolean success = false;
157. if(isLogined()){
158. //已經登入成功，開始傳送資料
159. channel = tcpClient.getChannel(Constants.TCP_ADDRESS, Constants.TCP_PORT);
160. if(null != channel && channel.isWritable()){
161. Message msg = buildSendVO(awsVo);
162. ChannelBuffer msgBuffer = buildMessage(msg);
163. channel.write(msgBuffer);
164. LOGGER.error("傳送--" + awsVo.getVehicleNo());
165. }else{
166. LONGINSTATUS = "";
167. }
168. }else if(LOGIN_FLAG == 0){
169. LOGIN_FLAG ++;
170. login2Gov();
171. LOGGER.error("--------------第一次登入");
172. }else{
173. LOGGER.error("--------------等待登入");
174. }
175. return success;
176. }
179. /**
180. * 轉換VO
181. * void
182. * @param awsVo
183. * 2016年9月28日 by fox_mt
184. */
185. private Message buildSendVO(Idc2AwsGpsVo awsVo){
186. Message msg = new Message(JT809Constants.UP_EXG_MSG);
187. ChannelBuffer buffer = ChannelBuffers.buffer(36);
188. //是否加密
189. buffer.writeByte((byte)0);//0未加密 // 1
190. //日月年dmyy
191. Calendar cal = Calendar.getInstance();
192. cal.setTime(new Date());
193. buffer.writeByte((byte)cal.get(Calendar.DATE));
194. buffer.writeByte((byte)(cal.get(Calendar.MONTH) + 1));
195. String hexYear = "0" + Integer.toHexString(cal.get(Calendar.YEAR));
196. buffer.writeBytes(hexStringToByte(hexYear));//4
198. //時分秒
199. buffer.writeByte((byte)cal.get(Calendar.HOUR_OF_DAY));
200. buffer.writeByte((byte)cal.get(Calendar.MINUTE));
201. buffer.writeByte((byte)cal.get(Calendar.SECOND));//3
202. //經度，緯度
203. buffer.writeInt(formatLonLat(awsVo.getLon()));//4
204. buffer.writeInt(formatLonLat(awsVo.getLat()));//4
205. //速度
206. buffer.writeShort(awsVo.getSpeed().shortValue());//2
207. //行駛記錄速度
208. buffer.writeShort(awsVo.getSpeed().shortValue());//2
209. //車輛當前總里程數
210. buffer.writeInt(awsVo.getMileage().intValue());//4
211. //方向
212. buffer.writeShort(awsVo.getDirection().shortValue());//2
213. //海拔
214. buffer.writeShort((short)0);//2
215. //車輛狀態
216. int accStatus = null == awsVo.getAcc() ? 0 : awsVo.getAcc();
217. int gpsStatus = null == awsVo.getGpsStatus() ? 0 : awsVo.getGpsStatus();
218. if(accStatus == 0 && gpsStatus == 0){
219. buffer.writeInt(0);//4
220. }else if(accStatus == 1 && gpsStatus == 0){
221. buffer.writeInt(1);//4
222. }else if(accStatus == 0 && gpsStatus == 1){
223. buffer.writeInt(2);//4
224. }else{
225. buffer.writeInt(3);//4
226. }
227. //報警狀態
228. buffer.writeInt(0);//0表示正常；1表示報警//4
229. ChannelBuffer headBuffer = ChannelBuffers.buffer(buffer.capacity() + 28);
230. headBuffer.writeBytes(getBytesWithLengthAfter(21 , awsVo.getVehicleNo().getBytes(Charset.forName("GBK"))));//21
231. headBuffer.writeByte((byte)InitVehicleData.vehicleColor(awsVo.getVehicleNo()));//1
232. headBuffer.writeShort(JT809Constants.UP_EXG_MSG_REAL_LOCATION);//2
233. headBuffer.writeInt(buffer.capacity());//4
234. headBuffer.writeBytes(buffer);
235. msg.setMsgBody(headBuffer);
237. return msg;
238. }
240. /**
241. * 報文轉義
242. * void
243. * @param bytes
244. * @param formatBuffer
245. * 2016年10月12日 by fox_mt
246. */
247. private static void formatBuffer(byte[] bytes, ChannelBuffer formatBuffer){
248. for (byte b : bytes) {
249. switch(b){
250. case 0x5b:
251. byte[] formatByte0x5b = new byte[2];
252. formatByte0x5b[0] = 0x5a;
253. formatByte0x5b[1] = 0x01;
254. formatBuffer.writeBytes(formatByte0x5b);
255. break;
256. case 0x5a:
257. byte[] formatByte0x5a = new byte[2];
258. formatByte0x5a[0] = 0x5a;
259. formatByte0x5a[1] = 0x02;
260. formatBuffer.writeBytes(formatByte0x5a);
261. break;
262. case 0x5d:
263. byte[] formatByte0x5d = new byte[2];
264. formatByte0x5d[0] = 0x5e;
265. formatByte0x5d[1] = 0x01;
266. formatBuffer.writeBytes(formatByte0x5d);
267. break;
268. case 0x5e:
269. byte[] formatByte0x5e = new byte[2];
270. formatByte0x5e[0] = 0x5e;
271. formatByte0x5e[1] = 0x02;
272. formatBuffer.writeBytes(formatByte0x5e);
273. break;
274. default: formatBuffer.writeByte(b);break;
275. }
276. }
277. }
279. /**
280. * 16進位制字串轉換成byte陣列
281. * byte[]
282. * @param hex
283. * @return
284. * 2016年10月12日 by fox_mt
285. */
286. public static byte[] hexStringToByte(String hex) {
287. hex = hex.toUpperCase();
288. int len = (hex.length() / 2);
289. byte[] result = new byte[len];
290. char[] achar = hex.toCharArray();
291. for (int i = 0; i < len; i++) {
292. int pos = i * 2;
293. result[i] = (byte) (toByte(achar[pos]) << 4 | toByte(achar[pos + 1]));
294. }
295. return result;
296. }
298. private static byte toByte(char c) {
299. byte b = (byte) "0123456789ABCDEF".indexOf(c);
300. return b;
301. }
303. /**
304. * 格式化經緯度,保留六位小數
305. * int
306. * @param needFormat
307. * @return
308. * 2016年10月12日 by fox_mt
309. */
310. private int formatLonLat(Double needFormat){
311. NumberFormat numFormat = NumberFormat.getInstance();
312. numFormat.setMaximumFractionDigits(6);
313. numFormat.setGroupingUsed(false);
314. String fristFromat = numFormat.format(needFormat);
315. Double formatedDouble = Double.parseDouble(fristFromat);
316. numFormat.setMaximumFractionDigits(0);
317. String formatedValue = numFormat.format(formatedDouble * 1000000);
318. return Integer.parseInt(formatedValue);
319. }
321. /**
322. * 補全位數不夠的定長引數
323. * byte[]
324. * @param length
325. * @param pwdByte
326. * @return
327. * 2016年10月12日 by fox_mt
328. */
329. private byte[] getBytesWithLengthAfter(int length, byte[] pwdByte){
330. byte[] lengthByte = new byte[length];
331. for(int i = 0; i < pwdByte.length; i ++){
332. lengthByte[i] = pwdByte[i];
333. }
334. for (int i = 0; i < (length- pwdByte.length); i++) {
335. lengthByte[pwdByte.length + i] = 0x00;
336. }
337. return lengthByte;
338. }
341. public static void main(String[] args) {
342. TcpClientDemo s = TcpClientDemo.getInstance();
343. Idc2AwsGpsVo awsVo = new Idc2AwsGpsVo();
344. awsVo.setDirection(120);
345. awsVo.setLon(117.2900911);
346. awsVo.setLat(39.56362);
347. awsVo.setSpeed(45D);
348. awsVo.setMileage(10001D);
349. awsVo.setVehicleNo("XXXXX");
350. s.sendMsg2Gov(awsVo);
351. try {
352. Thread.sleep(20*1000);
353. } catch (InterruptedException e) {
354. e.printStackTrace();
355. }
356. s.sendMsg2Gov(awsVo);
357. }
359. }

這裡其實也沒有太多好說的，整體說一下，先按照協議拼接資料，拼接完成後，計算CRC，進行轉義，由於協議規定開始標記為0x5b,結束標記為0x5d，資料中可能包涵這些標記，所以必須轉移，具體轉換規則，在協議中就能找到，有一點要注意，一定是先計算crc，再轉一，將否則無法通過crc校驗，我就是在這裡吃了虧，耗了一天，再別人的指點下才解決，協議中crc計算使用的是crc16-ccitt標準，這裡找了好多實現，每個計算結果都不一樣，我也沒有仔細研究，我用的是沒問題的，程式碼如下

3. public class JT809Constants {
5. public static int UP_CONNECT_REQ = 0x1001;//主鏈路登入請求訊息
7. public static int UP_CONNECT_RSP = 0x1002;//主鏈路登入應答訊息
8. public static int UP_CONNECT_RSP_SUCCESS = 0x00;//登入成功
9. public static int UP_CONNECT_RSP_ERROR_01 = 0x01;//IP 地址不正確
10. public static int UP_CONNECT_RSP_ERROR_02 = 0x02;//接入碼不正確
11. public static int UP_CONNECT_RSP_ERROR_03 = 0x03;//使用者沒註冊
12. public static int UP_CONNECT_RSP_ERROR_04 = 0x04;//密碼錯誤
13. public static int UP_CONNECT_RSP_ERROR_05 = 0x05;//資源緊張,稍後再連線(已經佔 用);
14. public static int UP_CONNECT_RSP_ERROR_06 = 0x06;//其他
16. public static int UP_DICONNECE_REQ = 0x1003;//主鏈路登出請求訊息
18. public static int UP_DISCONNECT_RSP = 0x1004;//主鏈路登出應答訊息
20. public static int UP_LINKETEST_REQ = 0x1005;//主鏈路連線保持請求訊息
22. public static int UP_LINKTEST_RSP = 0x1006;//主鏈路連線保持應答訊息
24. public static int UP_DISCONNECT_INFORM = 0x1007;//主鏈路斷開通知訊息
26. public static int UP_CLOSELINK_INFORM = 0x1008;//下級平臺主動關閉鏈路通 知訊息
28. public static int DOWN_CONNECT_REQ = 0x9001;//從鏈路連線請求訊息
30. public static int DOWN_CONNECT_RSP = 0x9002;//從鏈路連線應答訊息
32. public static int DOWN_DISCONNECT_REQ = 0x9003;//從鏈路登出請求訊息
34. public static int UP_EXG_MSG = 0x1200;//主鏈路動態資訊交換訊息
35. public static int UP_EXG_MSG_REAL_LOCATION = 0x1202;//實時上傳車輛定位資訊
36. public static int UP_EXG_MSG_HISTORY_LOCATION = 0x1203;//車輛定位資訊自動補報
38. public static int DOWN_EXG_MSG = 0x9200;//從鏈路動態資訊交換訊息
40. }
45. import java.io.Serializable;
46. import java.util.Arrays;
48. import org.jboss.netty.buffer.ChannelBuffer;
50. public class Message implements Serializable{
51. private static final long serialVersionUID = 4398559115325723920L;
53. public static final int MSG_HEAD = 0x5b;
54. public static final int MSG_TALL = 0x5d;
56. //報文中除資料體外，固定的資料長度
57. public static final int MSG_FIX_LENGTH = 26;
59. private static int internalMsgNo = 0;
60. private long msgLength;
61. private long encryptFlag=1;
62. private int msgGesscenterId;
63. private long encryptKey;
64. private int crcCode;
65. private int msgId;
66. private int msgSn;
68. private ChannelBuffer msgBody;
69. private byte[] versionFlag = {0,0,1};
71. //下行報文標識，值為1時，代表傳送的資料；預設為0，代表接收的報文
72. // private int downFlag = 0;
74. public Message(){}
77. public Message(int msgId){
78. //下行報文需要填充報文序列號
79. synchronized((Integer)internalMsgNo) {
80. if(internalMsgNo == Integer.MAX_VALUE){
81. internalMsgNo = 0;
82. }
83. }
84. this.msgSn = ++internalMsgNo;
85. this.msgId = msgId;
86. //this.downFlag = 1;
87. }
90. public static int getInternalMsgNo() {
91. return internalMsgNo;
92. }
95. public static void setInternalMsgNo(int internalMsgNo) {
96. Message.internalMsgNo = internalMsgNo;
97. }
100. public long getMsgLength() {
101. return msgLength;
102. }
105. public void setMsgLength(long msgLength) {
106. this.msgLength = msgLength;
107. }
110. public long getEncryptFlag() {
111. return encryptFlag;
112. }
115. public void setEncryptFlag(long encryptFlag) {
116. this.encryptFlag = encryptFlag;
117. }
120. public int getMsgGesscenterId() {
121. return msgGesscenterId;
122. }
125. public void setMsgGesscenterId(int msgGesscenterId) {
126. this.msgGesscenterId = msgGesscenterId;
127. }
128. public void setMsgGesscenterId(long msgGesscenterId) {
129. this.msgGesscenterId = (int)msgGesscenterId;
130. }
133. public long getEncryptKey() {
134. return encryptKey;
135. }
138. public void setEncryptKey(long encryptKey) {
139. this.encryptKey = encryptKey;
140. }
143. public int getCrcCode() {
144. return crcCode;
145. }
148. public void setCrcCode(int crcCode) {
149. this.crcCode = crcCode;
150. }
153. public int getMsgId() {
154. return msgId;
155. }
158. public void setMsgId(int msgId) {
159. this.msgId = msgId;
160. }
163. public int getMsgSn() {
164. return msgSn;
165. }
168. public void setMsgSn(int msgSn) {
169. this.msgSn = msgSn;
170. }
173. public ChannelBuffer getMsgBody() {
174. return msgBody;
175. }
178. public void setMsgBody(ChannelBuffer msgBody) {
179. this.msgBody = msgBody;
180. }
183. public byte[] getVersionFlag() {
184. return versionFlag;
185. }
188. public void setVersionFlag(byte[] versionFlag) {
189. this.versionFlag = versionFlag;
190. }
193. public static int getMsgHead() {
194. return MSG_HEAD;
195. }
198. public static int getMsgTall() {
199. return MSG_TALL;
200. }
203. public static int getMsgFixLength() {
204. return MSG_FIX_LENGTH;
205. }
208. @Override
209. public String toString() {
210. return "Message [msgLength=" + msgLength + ", encryptFlag=" + encryptFlag + ", msgGesscenterId="
211. + msgGesscenterId + ", encryptKey=" + encryptKey + ", crcCode=" + crcCode + ", msgId=" + msgId
212. + ", msgSn=" + msgSn + ", msgBody=" + msgBody + ", versionFlag=" + Arrays.toString(versionFlag) + "]";
213. }
215. }

下面是Decoder

3. import org.jboss.netty.buffer.ChannelBuffer;
4. import org.jboss.netty.channel.Channel;
5. import org.jboss.netty.channel.ChannelHandlerContext;
6. import org.jboss.netty.handler.codec.frame.FrameDecoder;
9. public class Decoder extends FrameDecoder {
12. @Override
13. protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer) throws Exception {
14. int head = buffer.getByte(0);
15. int tail = buffer.getByte(buffer.capacity() - 1);
16. if( !(head == Message.MSG_HEAD && tail == Message.MSG_TALL)){
17. return null;
18. }
19. buffer.skipBytes(1);
20. Message msg = this.buildMessage(buffer);
21. return msg;
22. }
24. private Message buildMessage(ChannelBuffer buffer){
25. Message msg = new Message();
26. msg.setMsgLength(buffer.readUnsignedInt());
27. msg.setMsgSn(buffer.readInt());//4byte
28. msg.setMsgId(buffer.readUnsignedShort());//2byte
29. msg.setMsgGesscenterId(buffer.readUnsignedInt());//4byte
30. msg.setVersionFlag(buffer.readBytes(3).array());//3byte
31. msg.setEncryptFlag(buffer.readUnsignedByte());//1byte
32. msg.setEncryptKey(buffer.readUnsignedInt());//4byte
33. if(buffer.readableBytes() >= 9){
34. buffer.skipBytes(buffer.readableBytes() - 8);
35. }
36. ChannelBuffer bodyBytes = buffer.readBytes(buffer.readableBytes() -3);
37. msg.setMsgBody(bodyBytes);
38. msg.setCrcCode(buffer.readUnsignedShort());//2byte
39. buffer.skipBytes(1);
40. return msg;
41. }
43. }

這個decoder值得說一下，有一些朋友應該能看出來

1. if(buffer.readableBytes() >= 9){
2. buffer.skipBytes(buffer.readableBytes() - 8);
3. }

這段程式碼很突兀，其實我也很鬱悶，登陸的時候收到的反饋資訊，資訊長度竟然不是一樣的，有的時候長度是32，有的時候長度是31，長度是32的時候，如果不跳過一位，那麼解析出來的反饋狀態碼就是錯誤的

下面是recevieHandler，

3. import org.jboss.netty.buffer.ChannelBuffer;
4. import org.jboss.netty.channel.ChannelHandlerContext;
5. import org.jboss.netty.channel.MessageEvent;
6. import org.jboss.netty.channel.SimpleChannelHandler;
7. import org.slf4j.Logger;
8. import org.slf4j.LoggerFactory;
11. public class RecevieHandler extends SimpleChannelHandler{
13. private static Logger LOG = LoggerFactory.getLogger(RecevieHandler.class);
16. @Override
17. public void messageReceived(ChannelHandlerContext ctx, MessageEvent e){
18. Message msg = (Message) e.getMessage();
19. LOG.error("應答----------------" + "0x" + Integer.toHexString(msg.getMsgId()));
20. if(msg.getMsgId() == JT809Constants.UP_CONNECT_RSP){
21. ChannelBuffer msgBody = msg.getMsgBody();
22. int result = msgBody.readByte();
23. if(result == JT809Constants.UP_CONNECT_RSP_SUCCESS){
24. TcpClientFuJian.LONGINSTATUS = Constants.LOGIN_SUCCESS;
25. LOG.error("------------------登入成功");
26. }else{
27. LOG.error("------------------登入異常，請檢查" + "0x0" + Integer.toHexString(result));
28. }
29. }
30. }
31. }

根據JT809協議，應該是一個主鏈路，一個從鏈路，從鏈路用來接收下發的訊息，實際上不需要的，推送資料的，在主鏈路登入時會有反饋，從鏈路可以忽略，但是從鏈路的ip和埠，是必填引數，隨便給就可以了，

上面經過decoder之後得到Message，receiveHandler就是來判斷是否登陸，登陸成功的話，就可以傳送資料

這裡涉及登陸時機的問題，先看下心跳handler，我們再說登陸問題

HeartBeatHandler如下：

3. import org.apache.commons.lang3.StringUtils;
4. import org.jboss.netty.channel.ChannelHandlerContext;
5. import org.jboss.netty.handler.timeout.IdleState;
6. import org.jboss.netty.handler.timeout.IdleStateAwareChannelHandler;
7. import org.jboss.netty.handler.timeout.IdleStateEvent;
8. import org.slf4j.Logger;
9. import org.slf4j.LoggerFactory;
11. import com.zuche.us.gpsidc2aws.constant.JT809Constants;
12. import com.zuche.us.gpsidc2aws.tcpclient.vo.Message;
14. public class HeartBeatHandler extends IdleStateAwareChannelHandler {
16. private static Logger LOG = LoggerFactory.getLogger(HeartBeatHandler.class);
17. @Override
18. public void channelIdle(ChannelHandlerContext ctx, IdleStateEvent e) throws Exception{
20. if(StringUtils.isBlank(TcpClientFuJian.LONGINSTATUS) || TcpClientFuJian.LOGINING.equals(TcpClientFuJian.LONGINSTATUS)){
21. TcpClientFuJian.getInstance().login2FuJianGov();
22. LOG.error("利用空閒心跳去登入------ 開始登入");
23. }
25. if(e.getState() == IdleState.WRITER_IDLE){
26. LOG.error("鏈路空閒，傳送心跳!");
27. Message msg = new Message(JT809Constants.UP_LINKETEST_REQ);
28. e.getChannel().write(TcpClientFuJian.buildMessage(msg));
29. super.channelIdle(ctx, e);
30. }
31. }
33. }

心跳包中包含了登陸，這個怎麼說呢，沒想到什麼好辦法進行登陸失敗重登陸情況，因為一個不小心，可能就會一直在登陸

所以在傳送資料時如果發現沒登陸，那麼進行登陸，通過flag來判斷，如果登陸過了，那就不管了，成功就發資料，失敗就等著，等到心跳包傳送心跳時去檢測是否登陸成功了，沒成功的話讓心跳包去再登陸一次，這樣省了不少程式碼，就是肯能比較混亂，這個看個人想法了

只要能實現就行，

最後時crc16工具類

3. public class CRC16CCITT {
5. public static int crc16(byte[] bytes){
6. int crc = 0xFFFF;
7. for (int j = 0; j < bytes.length ; j++) {
8. crc = ((crc >>> 8) | (crc << 8) )& 0xffff;
9. crc ^= (bytes[j] & 0xff);//byte to int, trunc sign
10. crc ^= ((crc & 0xff) >> 4);
11. crc ^= (crc << 12) & 0xffff;
12. crc ^= ((crc & 0xFF) << 5) & 0xffff;
13. }
14. crc &= 0xffff;
15. return crc;
16. }
21. }

我用的是直接計算的方式，還有一種查表法，這裡就不說了，百度下應該能找到

到此，客戶端實現就完成了

不過由於我們是多臺伺服器推送資料，所以必須建立多個tcp連結，接收端之前沒這麼收過資料，中間經歷了一點兒小波折，後臺接收端根據我們的伺服器數量給開通了多個賬號，

多個賬號就涉及到哪臺伺服器用哪個賬號的問題，所以就有了上面程式碼中的配置檔案

檔案內容很簡單

ip=使用者名稱：介入碼：密碼

例如:

127.0.0.1=101:101:666666

為每臺伺服器制定好所使用的賬號;

另外車牌顏色，顏色的狀態需要使用JT/T415-2006標準協議，這個挺難找的，我直接把顏色和狀態碼貼出來，方便大家參考

JT/T415-2006
1:藍色
2:黃色
3:黑色
4:白色
9:其他

至此整個客戶端完滿，經過測試，完美上線，沒有出現問題，有一點稍微注意下，程式碼中傳送資料的時候，時間的小時的12進位制的，看接收端的要求，可以改成24小時制的

修改方法很簡單，將Calendar.HOUR 改成Calendar.HOUR_OF_DAY 就可以

本地開發的時候，可能不知道傳送的資料對不對，這個對應的接收端應該都有解析工具，可以要來自己去校驗

差不多了再進行聯調

也可以自己模擬一個簡單的接收端，來進行測試，這個程式碼晚上有現成的，這裡就不多說了。

希望能幫助大家，如果又問題，可以隨時聯絡我