Netty自娛自樂之協議棧設計

---恢復內容開始---

　　俺工作已經一年又6個月了，想想過的真快，每天寫業務，寫業務，寫業務......。然后就是祈禱著，這次上線不要出現線上bug。繼續這每天無聊的增刪改查，學習學習一下自己感興趣的事，就把自己當作小學生。然后學學習，打發打發時間，如果以后自己能用到呢？這又有誰說的清楚。

?　　好了，最近在學習Netty，主要看了這2本書的一些內容，第一本就是《Netty實戰》，第二本就是《Netty權威指南》。然后在看到Netty權威指南上有一章比較感興趣，用了整整一章用來描寫如何取自己定義一個協議。接著閱讀完后，我就按照書本上的相關內容，去實現了一下。糾正了一下書本上的錯誤代碼。工作都是在開發電商項目，基本上對底層傳輸這一塊接觸甚少。如果有機會想去一個游戲公司，這樣看看能不能接觸更多的網絡傳輸相關內容。哎，不知道這樣的去轉有木有要，糾結。。。。。。。。。

　　好了，現在開始看書和事件的經歷吧。

　　現在，我們設計一個傳輸協議如下

2字節：協議固定值

1字節：主版本號 
1字節：副版本號

消息長度 ：消息頭 和消息體
4字節

回話ID, 全局唯一
8字節

 業務請求消息  
1:業務請求消息  
2:業務響應消息  
3:握手請求消息 
4：握手應答消息 
5：心跳請求消息  
6：心跳應答消息
1字節

優先級別
1字節

附件

code

length

sessionId

type

primary

attachment

　　上面的定義，是來著Netty的權威指南。這個是協議的頭。然后接下來是一個協議體。而協議體在編碼上就是一個Object.

協議頭	協議體
customHeader	bodyMessage

　　根據上面的定義，直接寫出協議定義model.直接上代碼：

 1 @Data
 2 @ToString
 3 public class NettyCustomHeader {
 4     /**
 5      * code 2字節：netty協議消息， 1字節：主版本號 1字節：副版本號  4
 6      */
 7     private int code = 0xABCD0101;
 8 
 9     /**
10      * 消息長度 ：消息頭 和消息題 32
11      */
12     private int length;
13 
14     /**
15      * 回話ID, 全局唯一 64
16      */
17     private long sessionId;
18 
19     /**
20      * 業務請求消息  1:業務請求消息  2:業務響應消息  3:握手請求消息 4：握手應答消息 5：心跳請求消息  6：心跳應答消息
21      */
22     private byte type;
23 
24     /**
25      * 優先級別
26      */
27     private byte primary;
28 
29     /**
30      * 附件
31      */
32     Map<String, Object> attachment;
33 
34 }

 1 @Data
 2 @ToString
 3 public class NettyCustomMessage {
 4 
 5     /**
 6      * 消息頭
 7      */
 8     private NettyCustomHeader customHeader;
 9 
10     /**
11      * 消息體
12      */
13     private Object bodyMessage;
14 
15 
16 }

　　學過Netty的同學或者了解的同學知道，Netty是通過ChannelHandler來處理IO消息的。我編碼的Netty版本是4。那么處理消息首先第一步就是解碼，LengthFieldBasedFrameDecoder這個解碼器是基于長度的解碼器，并且能解決TCP/IP包的粘包和拆包問題。代碼如下。

 1 public class ByteBuf2NettyMessageDecoder extends LengthFieldBasedFrameDecoder {
 2 
 3     // private NettyMarshallingDecoder marshallingDecoder = NettyMarshallingFactory.buildNettyMarshallingDecoder();
 4 
 5     public ByteBuf2NettyMessageDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength) {
 6         super(maxFrameLength, lengthFieldOffset, lengthFieldLength);
 7     }
 8 
 9     public ByteBuf2NettyMessageDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment, int initialBytesToStrip) {
10         super(maxFrameLength, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip);
11     }
12 
13     public ByteBuf2NettyMessageDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment, int initialBytesToStrip, boolean failFast) {
14         super(maxFrameLength, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip, failFast);
15     }
16 
17     public ByteBuf2NettyMessageDecoder(ByteOrder byteOrder, int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment, int initialBytesToStrip, boolean failFast) {
18         super(byteOrder, maxFrameLength, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip, failFast);
19     }
20 
21     @Override
22     protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
23         //調用父類decode ,得到整包消息
24         ByteBuf readBuf = (ByteBuf) super.decode(ctx, in);
25         if (readBuf == null) {
26             return null;
27         }
28         NettyCustomMessage customMessage = new NettyCustomMessage();
29         NettyCustomHeader customHeader = new NettyCustomHeader();
30         customHeader.setCode(readBuf.readInt());
31         customHeader.setLength(readBuf.readInt());
32         customHeader.setSessionId(readBuf.readLong());
33         customHeader.setType(readBuf.readByte());
34         customHeader.setPrimary(readBuf.readByte());
35 
36         int attachmentSize = readBuf.readByte();
37         if (attachmentSize > 0) {
38             Map<String, Object> attachment = new HashMap<String, Object>();
39             for (int i = 0; i < attachmentSize; i++) {
40                 int keySize = readBuf.readInt();
41                 byte[] keyByte = new byte[keySize];
42                 readBuf.readBytes(keyByte);
43                 String key = new String(keyByte, CharsetUtil.UTF_8.name());
44 
45                 Object value = JavaByteFactory.decode(readBuf);
46                 //Object value = marshallingDecoder.decode(ctx, readBuf);
47                 attachment.put(key, value);
48             }
49             customHeader.setAttachment(attachment);
50         }
51 
52         customMessage.setCustomHeader(customHeader);
53         if (readBuf.readableBytes() > 0) {
54             Object body = JavaByteFactory.decode(readBuf);
55             //Object body = marshallingDecoder.decode(ctx, readBuf);
56             customMessage.setBodyMessage(body);
57         }
58 
59         return customMessage;
60     }
61 
62     @Override
63     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
64         System.out.println(cause.getStackTrace());
65         cause.getStackTrace();
66         super.exceptionCaught(ctx, cause);
67     }
68 }

　　上面注釋的原因，marshallingDecoder不支持java7,所以我自己寫了一個編碼／解碼幫助類，就是前4個字節代表長度，后面是就是時間內容。從上面的代碼我們知道，就是把ByteBuf轉化為自己定義的協議對象。從上面的解碼上，可能有點模糊，但是從下面的如何編碼上，就可以知道為啥是這么解碼的。

 1 public class NettyMessage2ByteBufEncoder extends MessageToMessageEncoder<NettyCustomMessage> {
 2 
 3     private NettyMarshallingEncoder nettyMarshallingEncoder;
 4 
 5     public NettyMessage2ByteBufEncoder() {
 6         // this.nettyMarshallingEncoder = NettyMarshallingFactory.buildNettyMarshallingEncoder();
 7 
 8     }
 9 
10     protected void encode(ChannelHandlerContext ctx, NettyCustomMessage msg, List<Object> out) throws Exception {
11 
12         if (msg == null || msg.getCustomHeader() == null) {
13             throw new Exception("the encode message is null");
14         }
15 
16         ByteBuf sendBuf = Unpooled.buffer();
17         sendBuf.writeInt(msg.getCustomHeader().getCode());
18         sendBuf.writeInt(msg.getCustomHeader().getLength());
19         sendBuf.writeLong(msg.getCustomHeader().getSessionId());
20         sendBuf.writeByte(msg.getCustomHeader().getType());
21         sendBuf.writeByte(msg.getCustomHeader().getPrimary());
22 
23         //attachment ,
24 
25         if (msg.getCustomHeader().getAttachment() != null) {
26             sendBuf.writeByte(msg.getCustomHeader().getAttachment().size());
27             String key = null;
28             byte[] keyArray = null;
29             for (Map.Entry<String, Object> entryKey : msg.getCustomHeader().getAttachment().entrySet()) {
30                 key = entryKey.getKey();
31                 keyArray = key.getBytes(CharsetUtil.UTF_8.name());
32                 sendBuf.writeInt(keyArray.length);
33                 sendBuf.writeBytes(keyArray);
34                 ByteBuf value = JavaByteFactory.encode(entryKey.getValue());
35                 sendBuf.writeBytes(value);
36                 // nettyMarshallingEncoder.encode(ctx, entryKey.getValue(), sendBuf);
37             }
38         } else {
39             sendBuf.writeByte(0);
40         }
41 
42 
43         if (msg.getBodyMessage() != null) {
44             ByteBuf value = JavaByteFactory.encode(msg.getBodyMessage());
45             sendBuf.writeBytes(value);
46             //nettyMarshallingEncoder.encode(ctx, msg.getBodyMessage(), sendBuf);
47         }
48 
49         //在第5個字節開始的int 是長度，重新設置
50         sendBuf.setInt(4, sendBuf.readableBytes());
51 
52         out.add(sendBuf);
53     }
54 
55     @Override
56     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
57         System.out.println(cause.getStackTrace());
58         cause.getStackTrace();
59         super.exceptionCaught(ctx, cause);
60     }
61 }

　　從上面可以知道解碼，就是把自定義協議對象?NettyCustomMessage 通過自己的規則放到ByteBuf上。代碼比較簡單，不解釋。JavaByteFactory的代碼如下：

 1 public class JavaByteFactory {
 2 
 3 
 4     public static Object decode(ByteBuf byteBuf) {
 5         if (byteBuf == null || byteBuf.readableBytes() <= 0) {
 6             return null;
 7         }
 8         int valueSize = byteBuf.readInt();
 9         byte[] value = new byte[valueSize];
10         byteBuf.readBytes(value);
11 
12         ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(value);
13         ObjectInputStream inputStream = null;
14         try {
15             inputStream = new ObjectInputStream(byteArrayInputStream);
16             return inputStream.readObject();
17         } catch (IOException e) {
18             e.printStackTrace();
19         } catch (ClassNotFoundException e) {
20             e.printStackTrace();
21         }
22         return null;
23 
24 
25     }
26 
27     public static ByteBuf encode(Object object) {
28         if (object == null) {
29             return null;
30         }
31         ByteArrayOutputStream byteOutput = new ByteArrayOutputStream();
32         try {
33             ObjectOutputStream objectOutputStream = new ObjectOutputStream(byteOutput);
34             objectOutputStream.writeObject(object);
35             byte[] bytes = byteOutput.toByteArray();
36 
37             ByteBuf buffer = Unpooled.buffer(bytes.length + 4);
38             buffer.writeInt(bytes.length);
39             buffer.writeBytes(bytes);
40             return buffer;
41 
42         } catch (IOException e) {
43             e.printStackTrace();
44         }
45         return null;
46     }

　　編碼就是首選把Object 對象轉換了byte []數組，然后寫入4個字節為byte[]數組的長度，接著是數組的內容到ByteBuf對象上。相應的解碼就是先獲取4個字節，得到后面字節長度，接著讀取指定長度即可。

　　接著心跳和權限檢測都是在解碼器之后進行業務的處理。直接上代碼。

　　下面是權限認證的請求handler和響應handler.

 1 public class AuthorityCertificationRequestHanlder extends ChannelInboundHandlerAdapter {
 2 
 3     @Override
 4     public void channelActive(ChannelHandlerContext ctx) throws Exception {
 5         ctx.writeAndFlush(buildAuthorityCertificationMsg());
 6     }
 7 
 8     @Override
 9     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
10         NettyCustomMessage message = (NettyCustomMessage) msg;
11         if (message != null && message.getCustomHeader() != null && message.getCustomHeader().getType() == NettyMessageConstant.CUSTOMER_AUTH_CERTI_TYPE) {
12             byte authResult = (Byte) message.getBodyMessage();
13             if (authResult != (byte) 0) { //握手失敗。關閉鏈接
14                 ctx.close();
15                 return;
16             }
17             System.out.println("authority certification is success .....");
18             ctx.fireChannelRead(msg);
19         } else {
20             ctx.fireChannelRead(msg);
21         }
22 
23     }
24 
25     @Override
26     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
27         cause.getStackTrace();
28         ctx.channel().close();
29         System.out.println(cause.getStackTrace());
30         ctx.fireExceptionCaught(cause);
31     }
32 
33 
34     protected NettyCustomMessage buildAuthorityCertificationMsg() {
35         NettyCustomMessage message = new NettyCustomMessage();
36         NettyCustomHeader customHeader = new NettyCustomHeader();
37         customHeader.setType(NettyMessageConstant.CUSTOMER_AUTH_CERTI_TYPE);
38         message.setCustomHeader(customHeader);
39         return message;
40     }
41 
42 }

 1 public class AuthorityCertificationResponseHanlder extends ChannelInboundHandlerAdapter {
 2 
 3     private Map<String, Boolean> authority = new ConcurrentHashMap<String, Boolean>();
 4 
 5     private String[] ipList = new String[]{"127.0.0.1"};
 6 
 7     @Override
 8     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 9 
10         NettyCustomMessage customMessage = (NettyCustomMessage) msg;
11         NettyCustomMessage response;
12         if (customMessage.getCustomHeader() != null && customMessage.getCustomHeader().getType() == NettyMessageConstant.CUSTOMER_AUTH_CERTI_TYPE) {
13             String remoteAddress = ctx.channel().remoteAddress().toString();
14             if (authority.containsKey(remoteAddress)) { //重復登陸
15                 response = buildAuthorCertiResponseMessage((byte) -1);
16             } else {
17                 InetSocketAddress inetSocketAddress = (InetSocketAddress) ctx.channel().remoteAddress();
18                 boolean isAuth = false;
19                 for (String ip : ipList) {
20                     if (ip.equals(inetSocketAddress.getAddress().getHostAddress())) {
21                         isAuth = true;
22                         break;
23                     }
24                 }
25                 if (isAuth) {
26                     response = buildAuthorCertiResponseMessage((byte) 0);
27                     authority.put(remoteAddress, true);
28                 } else {
29                     response = buildAuthorCertiResponseMessage((byte) -1);
30                 }
31             }
32             System.out.println("the client [" + remoteAddress + "] is connecting ,status:" + response);
33             ctx.writeAndFlush(response);
34             return;
35         }
36         ctx.fireChannelRead(msg);
37     }
38 
39 
40     @Override
41     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
42         System.out.println(cause.getStackTrace());
43         cause.getStackTrace();
44         String remoteAddress = ctx.channel().remoteAddress().toString();
45         authority.remove(remoteAddress);
46         ctx.channel().close();
47         ctx.fireExceptionCaught(cause);
48     }
49 
50     private NettyCustomMessage buildAuthorCertiResponseMessage(byte body) {
51         NettyCustomMessage message = new NettyCustomMessage();
52         NettyCustomHeader customHeader = new NettyCustomHeader();
53         customHeader.setType(NettyMessageConstant.SERVER_AUTH_CERTI_TYPE);
54         message.setCustomHeader(customHeader);
55         message.setBodyMessage(body);
56         return message;
57     }
58 
59 }

　　下面是心跳檢測handler

 1 public class HeartBeatCheckRequestHandler extends ChannelInboundHandlerAdapter {
 2 
 3     private volatile ScheduledFuture<?> scheduledFuture;
 4 
 5     @Override
 6     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 7         NettyCustomMessage customMessage = (NettyCustomMessage) msg;
 8         if (customMessage.getCustomHeader() != null && customMessage.getCustomHeader().getType() == NettyMessageConstant.SERVER_AUTH_CERTI_TYPE) {
 9             scheduledFuture = ctx.executor().scheduleAtFixedRate(new HeartBeatCheckTask(ctx), 0, 5000, TimeUnit.MILLISECONDS);
10             System.out.println("the client [ " + ctx.channel().localAddress().toString() + " ] send heart beat ...........");
11         } else if (customMessage.getCustomHeader() != null && customMessage.getCustomHeader().getType() == NettyMessageConstant.HEART_BEAT_CHECK_PONG_TYPE) {
12             System.out.println("the client [ " + ctx.channel().localAddress().toString() + " ] recieve heart beat .............");
13         } else {
14             ctx.fireChannelRead(msg);
15         }
16 
17     }
18 
19     @Override
20     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
21         System.out.println(cause.getStackTrace());
22         cause.getStackTrace();
23         if (scheduledFuture != null) {
24             scheduledFuture.cancel(true);
25             scheduledFuture = null;
26         }
27         ctx.fireExceptionCaught(cause);
28     }
29 
30     class HeartBeatCheckTask implements Runnable {
31 
32         private ChannelHandlerContext context;
33 
34         public HeartBeatCheckTask(ChannelHandlerContext context) {
35             this.context = context;
36         }
37 
38         @Override
39         public void run() {
40             NettyCustomMessage customMessage = new NettyCustomMessage();
41             NettyCustomHeader customHeader = new NettyCustomHeader();
42             customHeader.setType(NettyMessageConstant.HEART_BEAT_CHECK_PING_TYPE);
43             customMessage.setCustomHeader(customHeader);
44             context.writeAndFlush(customMessage);
45             System.out.println("the client [ " + context.channel().localAddress().toString() + " ] send heart beat to server ....");
46 
47         }
48     }
49 }

 1 public class HeartBeatCheckResponseHandler extends ChannelInboundHandlerAdapter {
 2 
 3     @Override
 4     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 5         NettyCustomMessage customMessage = (NettyCustomMessage) msg;
 6         if (customMessage.getCustomHeader() != null && customMessage.getCustomHeader().getType() == NettyMessageConstant.HEART_BEAT_CHECK_PING_TYPE) {
 7             System.out.println("the server recieve the client [ " + ctx.channel().remoteAddress().toString() + " ] heart beat check package,");
 8 
 9             NettyCustomMessage sendPongMessage = new NettyCustomMessage();
10             NettyCustomHeader customHeader = new NettyCustomHeader();
11             customHeader.setType(NettyMessageConstant.HEART_BEAT_CHECK_PONG_TYPE);
12             sendPongMessage.setCustomHeader(customHeader);
13             ctx.writeAndFlush(customMessage);
14             return;
15         }
16         ctx.fireChannelRead(msg);
17     }
18 
19     @Override
20     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
21         System.out.println(cause.getStackTrace());
22         cause.getStackTrace();
23         super.exceptionCaught(ctx, cause);
24     }
25 
26     @Override
27     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
28         System.out.println("the client [ " + ctx.channel().remoteAddress().toString() + " ] is close ....,then close channel");
29         ctx.channel().close();
30     }
31 
32 
33 }

　　最后是我們的客戶端和服務端代碼，如下：

 1 public class NettyProtocalClient {
 2     private ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1);
 3 
 4     private Bootstrap bootstrap;
 5 
 6     private EventLoopGroup eventLoopGroup;
 7 
 8     private String host;
 9 
10     private int port;
11 
12     private int localPort;
13 
14     public NettyProtocalClient(String host, int port) {
15         this(7777, host, port);
16     }
17 
18     public NettyProtocalClient(int localPort, String host, int port) {
19         this.host = host;
20         this.port = port;
21         this.localPort = localPort;
22     }
23 
24     public void connect() throws InterruptedException {
25         try {
26             bootstrap = new Bootstrap();
27             eventLoopGroup = new NioEventLoopGroup();
28             bootstrap.group(eventLoopGroup)
29                     .channel(NioSocketChannel.class)
30                     .option(ChannelOption.TCP_NODELAY, true)
31                     .handler(new ChannelInitializer<io.netty.channel.Channel>() {
32                         @Override
33                         protected void initChannel(Channel ch) throws Exception {
34                             ch.pipeline()
35                                     .addLast("log", new LoggingHandler(LogLevel.INFO))
36                                     .addLast("decoder", new ByteBuf2NettyMessageDecoder(6 * 1024, 4, 4, -8, 0, true))
37                                     .addLast("encoder", new NettyMessage2ByteBufEncoder())
38                                     .addLast("timeout", new ReadTimeoutHandler(50))
39                                     .addLast("authority", new AuthorityCertificationRequestHanlder())
40                                     .addLast("hearbeat", new HeartBeatCheckRequestHandler());
41 
42 
43                         }
44                     });
45             ChannelFuture future = bootstrap.connect(new InetSocketAddress(host, port), new InetSocketAddress("127.0.0.1", localPort)).sync();
46             future.channel().closeFuture().sync();
47         } finally {
48             if (eventLoopGroup != null) {
49                 eventLoopGroup.shutdownGracefully().sync();
50             }
51             executorService.execute(new Runnable() {
52                 @Override
53                 public void run() {
54                     try {
55                         TimeUnit.SECONDS.sleep(5);
56                         connect();
57                     } catch (InterruptedException e) {
58                         e.printStackTrace();
59                     }
60                 }
61             });
62 
63         }
64     }
65 }

 1 public class NettyProtocalServer {
 2     private ServerBootstrap serverBootstrap;
 3 
 4     private EventLoopGroup boss;
 5 
 6     private EventLoopGroup worker;
 7 
 8     private String host;
 9 
10 
11     private int port;
12 
13     public NettyProtocalServer(String host, int port) {
14         this.host = host;
15         this.port = port;
16     }
17 
18     public void start() throws InterruptedException {
19         try {
20             serverBootstrap = new ServerBootstrap();
21             boss = new NioEventLoopGroup(1);
22             worker = new NioEventLoopGroup();
23 
24 
25             serverBootstrap.group(boss, worker)
26                     .channel(NioServerSocketChannel.class)
27                     .handler(new LoggingHandler(LogLevel.INFO))
28                     .option(ChannelOption.SO_BACKLOG, 1024)
29                     .childHandler(new ChannelInitializer<Channel>() {
30                         @Override
31                         protected void initChannel(Channel ch) throws Exception {
32                             ch.pipeline()
33                                     .addLast("log",new LoggingHandler(LogLevel.INFO))
34                                     .addLast("decoder", new ByteBuf2NettyMessageDecoder(6 * 1024, 4, 4, -8, 0, true))
35                                     .addLast("encoder", new NettyMessage2ByteBufEncoder())
36                                     .addLast("timeout", new ReadTimeoutHandler(50))
37                                     .addLast("authority", new AuthorityCertificationResponseHanlder())
38                                     .addLast("hearbeat", new HeartBeatCheckResponseHandler());
39 
40                         }
41                     });
42             ChannelFuture future = serverBootstrap.bind(new InetSocketAddress(host, port)).sync();
43             future.channel().closeFuture().sync();
44         } finally {
45             if (boss != null) {
46                 boss.shutdownGracefully();
47             }
48             if (worker != null) {
49                 worker.shutdownGracefully();
50             }
51         }
52     }
53 }