Netty自娛自樂之協議棧設計
阿新 • • 發佈：2019-01-16
---恢復內容開始---
俺工作已經一年又6個月了，想想過的真快，每天寫業務，寫業務，寫業務......。然後就是祈禱著，這次上線不要出現線上bug。繼續這每天無聊的增刪改查，學習學習一下自己感興趣的事，就把自己當作小學生。然後學學習，打發打發時間，如果以後自己能用到呢？這又有誰說的清楚。
好了，最近在學習Netty，主要看了這2本書的一些內容，第一本就是《Netty實戰》，第二本就是《Netty權威指南》。然後在看到Netty權威指南上有一章比較感興趣，用了整整一章用來描寫如何取自己定義一個協議。接著閱讀完後，我就按照書本上的相關內容，去實現了一下。糾正了一下書本上的錯誤程式碼。工作都是在開發電商專案，基本上對底層傳輸這一塊接觸甚少。如果有機會想去一個遊戲公司，這樣看看能不能接觸更多的網路傳輸相關內容。哎，不知道這樣的去轉有木有要，糾結。。。。。。。。。
好了，現在開始看書和事件的經歷吧。
現在，我們設計一個傳輸協議如下
2位元組：協議固定值 1位元組：主版本號 |
訊息長度 ：訊息頭 和訊息體 |
回話ID, 全域性唯一 |
業務請求訊息 |
優先級別 |
附件 |
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 privateint 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