所谓的协定，是由语法、语义、时序这三个因素组成的一种标准，通信单方依照该协定标准来实现网络数据传输，这样通信单方能力实现数据失常通信和解析。
因为不同的中间件在性能方面有肯定差别，所以其实应该是没有一种标准化协定来满足不同差异化需要，因而很多中间件都会定义本人的通信协议，另外通信协议能够解决粘包和拆包问题。
在本篇文章中，咱们来实现一个自定义音讯协定。
自定义协定的因素
自定义协定，那这个协定必须要有组成的元素，
- 魔数：用来判断数据包的有效性
- 版本号：能够反对协定降级
- 序列化算法：音讯注释采纳什么样的序列化和反序列化形式，比方json、protobuf、hessian等
- 指令类型：也就是以后发送的是一个什么类型的音讯，像zookeeper中，它传递了一个Type
- 申请序号：基于双工协定，提供异步能力，也就是收到的异步音讯须要找到后面的通信申请进行响应解决
- 音讯长度
- 音讯注释
协定定义
sessionId | reqType | Content-Length | Content |其中Version,Content-Length,SessionId就是Header信息，Content就是交互的主体。
定义我的项目构造以及引入包
<dependency> <groupId>io.netty</groupId> <artifactId>netty-all</artifactId></dependency><dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-log4j12</artifactId></dependency><dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId></dependency>我的项目构造如图4-1所示：
- netty-message-mic : 示意协定模块。
- netty-message-server ：示意nettyserver。
图4-1
- 引入log4j.properties
在nettyMessage-mic中，包的构造如下。
定义Header
示意音讯头
@Datapublic class Header{ private long sessionId; //会话id : 占8个字节 private byte type; //音讯类型：占1个字节 private int length; //音讯长度 : 占4个字节}定义MessageRecord
示意音讯体
@Datapublic class MessageRecord{ private Header header; private Object body;}OpCode
定义操作类型
public enum OpCode { BUSI_REQ((byte)0), BUSI_RESP((byte)1), PING((byte)3), PONG((byte)4); private byte code; private OpCode(byte code) { this.code=code; } public byte code(){ return this.code; }}定义编解码器
别离定义对该音讯协定的编解码器
MessageRecordEncoder
@Slf4jpublic class MessageRecordEncoder extends MessageToByteEncoder<MessageRecord> { @Override protected void encode(ChannelHandlerContext channelHandlerContext, MessageRecord record, ByteBuf byteBuf) throws Exception { log.info("===========开始编码Header局部==========="); Header header=record.getHeader(); byteBuf.writeLong(header.getSessionId()); //保留8个字节的sessionId byteBuf.writeByte(header.getType()); //写入1个字节的申请类型 log.info("===========开始编码Body局部==========="); Object body=record.getBody(); if(body!=null){ ByteArrayOutputStream bos=new ByteArrayOutputStream(); ObjectOutputStream oos=new ObjectOutputStream(bos); oos.writeObject(body); byte[] bytes=bos.toByteArray(); byteBuf.writeInt(bytes.length); //写入音讯体长度:占4个字节 byteBuf.writeBytes(bytes); //写入音讯体内容 }else{ byteBuf.writeInt(0); //写入音讯长度占4个字节，长度为0 } }}MessageRecordDecode
@Slf4jpublic class MessageRecordDecode extends ByteToMessageDecoder { @Override protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception { MessageRecord record=new MessageRecord(); Header header=new Header(); header.setSessionId(byteBuf.readLong()); //读取8个字节的sessionid header.setType(byteBuf.readByte()); //读取一个字节的操作类型 record.setHeader(header); //如果byteBuf剩下的长度还有大于4个字节，阐明body不为空 if(byteBuf.readableBytes()>4){ int length=byteBuf.readInt(); //读取四个字节的长度 header.setLength(length); byte[] contents=new byte[length]; byteBuf.readBytes(contents,0,length); ByteArrayInputStream bis=new ByteArrayInputStream(contents); ObjectInputStream ois=new ObjectInputStream(bis); record.setBody(ois.readObject()); list.add(record); log.info("序列化进去的后果："+record); }else{ log.error("音讯内容为空"); } }}测试协定的解析和编码
EmbeddedChannel是netty专门改良针对ChannelHandler的单元测试而提供的
public class CodesMainTest { public static void main( String[] args ) throws Exception { EmbeddedChannel channel=new EmbeddedChannel( new LoggingHandler(), new MessageRecordEncoder(), new MessageRecordDecode()); Header header=new Header(); header.setSessionId(123456); header.setType(OpCode.PING.code()); MessageRecord record=new MessageRecord(); record.setBody("Hello World"); record.setHeader(header); channel.writeOutbound(record); ByteBuf buf= ByteBufAllocator.DEFAULT.buffer(); new MessageRecordEncoder().encode(null,record,buf); channel.writeInbound(buf); }}编码包剖析
运行上述代码后，会失去上面的一个信息
+-------------------------------------------------+ | 0 1 2 3 4 5 6 7 8 9 a b c d e f |+--------+-------------------------------------------------+----------------+|00000000| 00 00 00 00 00 01 e2 40 03 00 00 00 12 ac ed 00 |.......@........||00000010| 05 74 00 0b 48 65 6c 6c 6f 20 57 6f 72 6c 64 |.t..Hello World |+--------+-------------------------------------------------+----------------+依照协定标准：
- 后面8个字节示意sessionId
- 一个字节示意申请类型
- 4个字节示意长度
- 前面局部内容示意音讯体
测试粘包和半包问题
通过slice办法进行拆分，失去两个包。
ByteBuf中提供了一个slice办法，这个办法能够在不做数据拷贝的状况下对原始ByteBuf进行拆分。
public class CodesMainTest { public static void main( String[] args ) throws Exception { //EmbeddedChannel是netty专门针对ChannelHandler的单元测试而提供的类。能够通过这个类来测试channel输出入站和出站的实现 EmbeddedChannel channel=new EmbeddedChannel( //解决粘包和半包问题// new LengthFieldBasedFrameDecoder(2048,10,4,0,0), new LoggingHandler(), new MessageRecordEncoder(), new MessageRecordDecode()); Header header=new Header(); header.setSessionId(123456); header.setType(OpCode.PING.code()); MessageRecord record=new MessageRecord(); record.setBody("Hello World"); record.setHeader(header); channel.writeOutbound(record); ByteBuf buf= ByteBufAllocator.DEFAULT.buffer(); new MessageRecordEncoder().encode(null,record,buf); //*********模仿半包和粘包问题************// //把一个包通过slice拆分成两个局部 ByteBuf bb1=buf.slice(0,7); //获取后面7个字节 ByteBuf bb2=buf.slice(7,buf.readableBytes()-7); //获取前面的字节 bb1.retain(); channel.writeInbound(bb1); channel.writeInbound(bb2); }}运行上述代码会失去如下异样， readerIndex(0) +length(8)示意要读取8个字节，然而只收到7个字节，所以间接报错。
2021-08-31 15:53:01,385 [io.netty.handler.logging.LoggingHandler]-[DEBUG] [id: 0xembedded, L:embedded - R:embedded] READ: 7B +-------------------------------------------------+ | 0 1 2 3 4 5 6 7 8 9 a b c d e f |+--------+-------------------------------------------------+----------------+|00000000| 00 00 00 00 00 01 e2 |....... |+--------+-------------------------------------------------+----------------+2021-08-31 15:53:01,397 [io.netty.handler.logging.LoggingHandler]-[DEBUG] [id: 0xembedded, L:embedded - R:embedded] READ COMPLETEException in thread "main" io.netty.handler.codec.DecoderException: java.lang.IndexOutOfBoundsException: readerIndex(0) + length(8) exceeds writerIndex(7): UnpooledSlicedByteBuf(ridx: 0, widx: 7, cap: 7/7, unwrapped: PooledUnsafeDirectByteBuf(ridx: 0, widx: 31, cap: 256))解决拆包问题
LengthFieldBasedFrameDecoder是长度域解码器，它是解决拆包粘包最罕用的解码器，基本上能笼罩大部分基于长度拆包的场景。其中开源的消息中间件RocketMQ就是应用该解码器进行解码的。
首先来阐明一下该解码器的外围参数
- lengthFieldOffset，长度字段的偏移量，也就是寄存长度数据的起始地位
- lengthFieldLength，长度字段锁占用的字节数
- lengthAdjustment，在一些较为简单的协定设计中，长度域不仅仅蕴含音讯的长度，还蕴含其余数据比方版本号、数据类型、数据状态等，这个时候咱们能够应用lengthAdjustment进行修改，它的值=包体的长度值-长度域的值
- initialBytesToStrip，解码后须要跳过的初始字节数，也就是音讯内容字段的起始地位
- lengthFieldEndOffset，长度字段完结的偏移量， 该属性的值=lengthFieldOffset+lengthFieldLength
public class CodesMainTest { public static void main( String[] args ) throws Exception { EmbeddedChannel channel=new EmbeddedChannel( //解决粘包和半包问题 new LengthFieldBasedFrameDecoder(1024, 9,4,0,0), new LoggingHandler(), new MessageRecordEncoder(), new MessageRecordDecode()); Header header=new Header(); header.setSessionId(123456); header.setType(OpCode.PING.code()); MessageRecord record=new MessageRecord(); record.setBody("Hello World"); record.setHeader(header); channel.writeOutbound(record); ByteBuf buf= ByteBufAllocator.DEFAULT.buffer(); new MessageRecordEncoder().encode(null,record,buf); //*********模仿半包和粘包问题************// //把一个包通过slice拆分成两个局部 ByteBuf bb1=buf.slice(0,7); ByteBuf bb2=buf.slice(7,buf.readableBytes()-7); bb1.retain(); channel.writeInbound(bb1); channel.writeInbound(bb2); }}增加一个长度解码器，就解决了拆包带来的问题。运行后果如下
2021-08-31 16:09:35,115 [com.netty.example.codec.MessageRecordDecode]-[INFO] 序列化进去的后果：MessageRecord(header=Header(sessionId=123456, type=3, length=18), body=Hello World)2021-08-31 16:09:35,116 [io.netty.handler.logging.LoggingHandler]-[DEBUG] [id: 0xembedded, L:embedded - R:embedded] READ COMPLETE基于自定义音讯协定通信
上面咱们把整个通信过程编写残缺，代码构造如图4-2所示.
图4-2
服务端开发
@Slf4jpublic class ProtocolServer { public static void main(String[] args){ EventLoopGroup boss = new NioEventLoopGroup(); //2 用于对承受客户端连贯读写操作的线程工作组 EventLoopGroup work = new NioEventLoopGroup(); ServerBootstrap b = new ServerBootstrap(); b.group(boss, work) //绑定两个工作线程组 .channel(NioServerSocketChannel.class) //设置NIO的模式 // 初始化绑定服务通道 .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel sc) throws Exception { sc.pipeline() .addLast( new LengthFieldBasedFrameDecoder(1024, 9,4,0,0)) .addLast(new MessageRecordEncoder()) .addLast(new MessageRecordDecode()) .addLast(new ServerHandler()); } }); ChannelFuture cf= null; try { cf = b.bind(8080).sync(); log.info("ProtocolServer start success"); cf.channel().closeFuture().sync(); } catch (InterruptedException e) { e.printStackTrace(); }finally { work.shutdownGracefully(); boss.shutdownGracefully(); } }}ServerHandler
@Slf4jpublic class ServerHandler extends ChannelInboundHandlerAdapter { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { MessageRecord messageRecord=(MessageRecord)msg; log.info("server receive message:"+messageRecord); MessageRecord res=new MessageRecord(); Header header=new Header(); header.setSessionId(messageRecord.getHeader().getSessionId()); header.setType(OpCode.BUSI_RESP.code()); String message="Server Response Message!"; res.setBody(message); header.setLength(message.length()); ctx.writeAndFlush(res); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { log.error("服务器读取数据异样"); super.exceptionCaught(ctx, cause); ctx.close(); }}客户端开发
public class ProtocolClient { public static void main(String[] args) { //创立工作线程组 EventLoopGroup group = new NioEventLoopGroup(); Bootstrap b = new Bootstrap(); b.group(group).channel(NioSocketChannel.class) .handler(new ChannelInitializer<SocketChannel>() { @Override public void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024, 9,4,0,0)) .addLast(new MessageRecordEncoder()) .addLast(new MessageRecordDecode()) .addLast(new ClientHandler()); } }); // 发动异步连贯操作 try { ChannelFuture future = b.connect(new InetSocketAddress("localhost", 8080)).sync(); Channel c = future.channel(); for (int i = 0; i < 500; i++) { MessageRecord message = new MessageRecord(); Header header = new Header(); header.setSessionId(10001); header.setType((byte) OpCode.BUSI_REQ.code()); message.setHeader(header); String context="我是申请数据"+i; header.setLength(context.length()); message.setBody(context); c.writeAndFlush(message); } //closeFuture().sync()就是让以后线程(即主线程)同步期待Netty server的close事件，Netty server的channel close后，主线程才会持续往下执行。closeFuture()在channel close的时候会告诉以后线程。 future.channel().closeFuture().sync(); } catch (InterruptedException e) { e.printStackTrace(); }finally { group.shutdownGracefully(); } }}ClientHandler
@Slf4jpublic class ClientHandler extends ChannelInboundHandlerAdapter { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { MessageRecord record=(MessageRecord)msg; log.info("Client Receive message:"+record); super.channelRead(ctx, msg); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { super.exceptionCaught(ctx, cause); ctx.close(); }}爆文举荐浏览：