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 }