网站域名空间地址,云南网站建设维修公司,长沙有什么互联网公司,公众号怎么做链接基于TCP的简单Netty自定义协议实现#xff08;万字#xff0c;全篇例子#xff09;
前言 有一阵子没写博客了#xff0c;最近在学习Netty写一个实时聊天软件#xff0c;一个高性能异步事件驱动的网络应用框架#xff0c;我们常用的SpringBoot一般基于Http协议#xff0…基于TCP的简单Netty自定义协议实现万字全篇例子
前言 有一阵子没写博客了最近在学习Netty写一个实时聊天软件一个高性能异步事件驱动的网络应用框架我们常用的SpringBoot一般基于Http协议而Netty是没有十分明确的协议的不过它内置了一些常用的通信协议当然你也可以自定义协议。 一、要求 接下来的内容默认你已经有了最基本的Java、Netty、Nio知识如果还没有这方面的知识的话可以先去小破站找个视频学习学习。 二、通信协议 * 本文提到的通信协议都是指基于TCP的应用层通信协议请勿理解错误。 1、协议基本单位 当数据在两台计算机上传输时传输的数据以比特(Bit)为单位就像01010100010010101...这种但是以比特作为传输单位太过精细、太过底层所以封装一下它将8个bit封装成一个单位就成了字节(Byte)所以一个协议的基本单位是字节Byte。同样的因为字节是其他大多数高级数据类型的基本组成所以通信协议的基本单位是字节。例如一串字节流可以被解析为视频、图片、字符串等等它是通用的。 也就是说我们要自定义一个通信协议就必须得自己解析字节。在SpringBoot框架中我们在Controller中能够直接得到字符串、对象的原因是框架已经帮我们将字节解析好了我们直接用就行但是如果我们要自定义协议就必须自力更生自己定义格式并解析它。 2、协议格式 协议的格式不是固定的协议只能是一个约定而不是强制要求。 举个例子假如你在晚自习上睡觉你提前和同桌约定好老师来了他就敲两下桌子班长来了他就敲三下桌子那么这种约定就可以认定为是一个通信协议但其并不是固定的因为明晚、后晚…你可以约定其他方式例如敲一下变成老师来了敲两下变成班长来了踢你一下表示老师来了踢你两下表示班长来了。并不是固定的。 基于这种思想我们可以定义一个简单的通信协议版本号为V1 请求地址 客户端IP 请求正文基于这个协议假如我们有一个请求它请求服务器的/test地址客户端IP是192.168.1.2请求正文是hello那么这个协议看起来就像 /test192.168.1.2hello将它转为字节流就是没有空格空格只是为了方便查看加的 47 116 101 115 116 49 57 50 46 49 54 56 46 49 46 50 104 101 108 108 111服务器在解析时就可以解析[0,5]个字符串为请求地址[/test]解析[6,11]个字符串为客户端IP[192.168.1.2]解析剩下的所有字符串为请求正文。 当然为了形象一点举了一个不太恰当的简单例子解析的不是字符串而是字节。 3、TCP的粘包半包
Ⅰ、问题描述 这个问题可能我一时半会解释不清楚导致粘包半包的原因很多感兴趣的可以去找找资料。 你只用知道基于TCP时数据并不是一次性达到的而是分段到达的例如我们上面举的例子那个协议数据/test192.168.1.2hello服务器在接收这些数据时它就有可能 第一次收到/test19
第二次收到2.168.
第三次收到1.2hello
...它可能不会一次收全可能要好几次所以我们上面定义的简单的协议就有一个问题它没有消息边界就是当客户端多次发送数据时服务器无法知道哪些数据是哪次请求的。还是刚才的例子 第一次收到/test192.168.1.2he
第二次收到llo/haha192.168.1.2hi在这两次数据中客户端分别发送了两次请求/test192.168.1.2hello 和 /haha192.168.1.2hi但是因为粘包半包的问题服务器不知道哪条是哪条了就会导致解析出错。 Ⅱ、如何解决 解决这个问题有很多种方法常见的方法有分隔符、标识请求长度等等。两种方法我都举个例子你也可以自己想一个方法来解决都是灵活的解决方法不是固定的。 分隔符的方法也很简单我们在每次请求结束时都添加一个特殊符号用于标识这个请求结束了服务器在解析时遇到这个特殊符号就知道这个请求结束了后面的数据是新请求的了。例如我们以$为分隔符服务器 第一次收到/test192.168.1.2
第二次收到hello$/haha192.168.1.2hi服务器在解析到$符时就知道/test请求已经结束了后面的数据是属于/haha请求的了。但是这么做的话有一个缺点就是之后传输的正文数据中不能含有$符不然解析依旧出错你也可以定义复杂一点的符号例如几个符号拼接也行$...。不过我要说的是其实你还可以用标识请求长度的方式解决。 标识请求长度就是客户端在传输请求之前先计算好整个请求有多少个字符为了不复杂先说成字符吧其实是字节再传输数据服务器在接收到数据后会去读取这个字段查看整个请求有多少个字符然后再根据这个数字读取多少个字符。那这就需要一个字段用来专门存储长度了。 基于这个需求我们上面定义的协议就得小小的升级一下变成V2 请求长度 请求地址 客户端IP 请求正文以后服务器会先读取开头的长度再根据长度读取后面的数据例如我们还是刚才的/test请求那么它将会变成 21/test192.168.1.2hello因为 /test192.168.1.2hello 总共是21个字符所以一开始就变为了21服务器一读取到开头的数字21就往下读取21个字符读完后就默认这个请求已经结束了再往下的就是其他请求了。 当然你也可以将长度字段包含在内那就是 23/test192.168.1.2hello这个长度可以出现在整个请求体的任何地方除了正文只要你在服务器/客户端解析的时候对应解析就行了。 暂时就介绍这个两个简单的方法其他的方法你可以自己想想出来了可以自己实现原则是能解决问题就是好办法。 三、创建协议
1、改正上面的说法 在上面的各个例子中我为了例子不复杂说的是解析字符其实解析的是字节(Byte)。 字符是字符字节是字节它们不一样你 是一个字符你好 是一个字符串而 -28(十进制) 它是一个字节-28、-67、-96 它们三个字节组成了一个字符 你 。 *** 在UTF8编码下常见的中文字符一般由3个字节组成不常见的一般是4个字节组成。 *** 在UTF8编码下英文字符一般由1个字节组成。 *** 数字的情况稍微复杂 1、8位的数字一般占用1字节范围从 -128 到 127 2、16位的数字一般占2字节范围从 -32,768 到 32,767 3、32位数字一般占4字节范围从 -2,147,483,648 到 2,147,483,647 4、64位数字一般占8字节范围从 -9,223,372,036,854,775,808 到 9,223,372,036,854,775,807 5、128位数字一般占16字节范围很大不写了。 例如在Rust中i32占4字节它对应的Java数字类型是inti64占8字节对应的Java类型是long以此类推。 JavaScript的number类型是64位的占8字节所以js要想表达64位以下的就有点麻烦了。 2、SP协议 解释了上面的错误后可以开始正式自定义协议了给这个协议取个名字就叫SP协议吧Simple Protocol译为简单的协议。 Ⅰ、报文长度 首先粘包半包的问题用长度字段解决4个字节表示的32位数字就够用了它的范围是-2,147,483,648 到 2,147,483,647负的20亿到正的20亿用来表示数据的话不算负数2147483648 / (1024 * 1024) 2048 MB也就是说32位数字所表示的数字范围正数用来表示数据大小的话可以表示2GB的数据一个请求根本不可能达到这么大所以32位的数字够用。因为2147483648个字节就是2GB。 那么协议开头就是 长度4字节Ⅱ、魔数 在协议中添加一个魔数用来标识这个报文是属于SP协议的服务器在网络中读取字节流时如果在长度字节后没有找到这个魔数就证明该字节流不是SP协议的就可以停止读取接下来的数据了可以做关闭连接、丢弃数据等操作就好像你去坐火车去北京火车进站时你看第二节车厢上有没有写目的地北京如果写了那么就是你要坐的火车如果没写那就证明不是你要坐的火车你可以等下一趟。其实就是为整个协议打一个标记。 魔数用几个字节都行为了不重复建议使用4字节的32位数字那么协议的第二部分应该是 长度4字节 魔数4字节Ⅲ、客户端身份 在多个客户端连接时服务器需要为每个客户端颁发一个标识用来区分不同的客户端的请求用几个字节都行为了不重复建议使用32字节的uuid作为客户端唯一标识。 那么协议第三部分是 长度4字节 魔数4字节 客户端标识32位Ⅳ、请求路径 请求路径这块比较灵活你可以使用1字节的8位数字表示也就是-128 到 127个数字。例如你可以规定1就是登录2就是注册等等。 我使用的是英文字符串的方式也就是一个字符一个字节但是路径长度不是不变的它会变化。例如 /test是5个字节但是 /hi 是3个字节不能像刚才一样用固定的长度来标识那么就需要一个固定的路径长度字段用来表示后续路径的长度。 于是协议的第四部分就是 长度4字节 魔数4字节 客户端标识32位 路径长度4字节 路径N字节Ⅴ、请求正文 到这步后这个简单的协议就基本完成了后续的正文长度是不定的但是我们有开头的长度字段表示整个报文的长度所以这个协议第五部分就是 长度4字节 魔数4字节 客户端标识32位 路径长度4字节 路径N字节 正文N字节3、完整协议 协议定义到这后基本完成了但是这只是一个简单的例子实际应用中肯定要复杂许多。 基于该协议模拟一个请求它请求/test路径使用Java字节码文件同款的魔数0xCAFEBABE请求正文是hello那么这个协议组装完成应该是这样的 | 54 | 0xCAFEBABE | 32位的UUID | 5 | /test | hello |解释一下首先魔数占了4字节UUID占了32字节路径长度占了4字节路径占了5字节正文占了5字节报文长度字段不计算在内所以总长度是4 32 4 5 5 54字节这就是开头54的由来。 路径 /test 前的 5 就是表示 /test 所占的 5 字节。 至此协议定义完成任何只要遵守了这个协议的请求都能够被Netty服务器识别。 四、服务器代码实现 协议定义好了该写服务器代码实现这个协议了。 1、Netty服务器启动流程 首先得先来复习一下Netty的启动流程我们才知道如何实现这个协议。 快速启动一个Netty服务器代码 public static void main(String[] args) {NioEventLoopGroup boss new NioEventLoopGroup(1);// 处理连接NioEventLoopGroup worker new NioEventLoopGroup();// 处理业务try {ChannelFuture channelFuture new ServerBootstrap().group(boss, worker) // 设置线程组.channel(NioServerSocketChannel.class) // 使用NIO通信模式.childHandler(new ChannelInitializerSocketChannel() {Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {// 在这里添加自定义的处理器}}).bind(8080).sync();// 绑定端口并启动服务器System.out.println(Netty Server is starting...);channelFuture.channel().closeFuture().sync();// 监听关闭} catch (InterruptedException e) {throw new RuntimeException(e);}finally {// 优雅的关闭线程组boss.shutdownGracefully();worker.shutdownGracefully();}
}要想自定义一个协议我们的重点在 initChannel() 方法上它可以为Netty添加处理器在TCP收到的数据传过来的时候处理原始的字节流数据 2、添加自定义处理器
Ⅰ、解释ChannelInitializer的作用 为了启动看起来清爽我们可以将childHandler()所需的参数抽取出来 public class CustomHandler extends ChannelInitializerSocketChannel {Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {}
}在childHandler()中传递.childHandler(new CustomHandler()) 原始的字节流数据在达到Netty的时候Netty内部会在我们自定义的处理器之前先做一些处理比如说将字节流数据封装成ByteBuf对象等等就像SprinigBoot我们添加自定义拦截器一样在我们添加的拦截器之前SpringBoot就已经添加了许多内部的拦截器先一步处理过数据了。 也就是说我们自定义处理器接收到的数据其实是经过ByteBuf封装过的字节流缓冲对象ByteBuf对象其实就是对Java.Nio中ByteBuffer的进一步封装升级。 画个简陋的图自定义处理器处理数据的整个流程看起来像这样 我们刚刚自定义的处理器初始化器就是这部分 它的作用就是往处理器链中添加一个个的自定义处理器在ChannelInitializer中添加处理器也很简单继承ChannelInitializer并实现它的initChannel方法再通过initChannel的形参SocketChannel获取到ChannelPipeline就可以添加了代码像这样 Override
protected void initChannel(SocketChannel channel) throws Exception {ChannelPipeline pipeline channel.pipeline();pipeline.addLast(处理器对象);// 添加一个个的处理器pipeline.addLast(处理器对象);// 添加一个个的处理器...
}Ⅱ、出站(Outbound)和入站(Inbound) 我没打错字是出站和入站不是出栈、入栈说白了其实就是数据进入Netty和数据从Netty发出进入Netty的行为叫入站Netty往外发送数据的行为叫出站。 所以处理器可以分为三种入站处理器、出站处理器、入站出站处理器入站处理器专门处理进入Netty的数据出站处理器专门处理从Netty发送的数据而入站出站处理器则两者都可以。 这些处理器看起来像这样 *** 注意出站处理器的顺序是与入站相反的出站是从尾巴上为第1个处理器头为最后一个处理器处理数据时会按照顺序一个一个进行。 有一个比喻可以很好理解它们之间的关系 处理器链pipeline就像两条相反的流水线pipeline.addLast();方法就像在流水线上安排一个工人调用一次就安排一个工人只不过一些工人专门处理过来的货物一些工人专门处理过去的货物。 好了接下来我们开始代码实现处理器了。 Ⅲ、处理器实现
①、处理长度 报文长度字段是我们自定义协议SP协议的第一个字段所以第一个处理器我们先处理长度。 首先这个处理器肯定是入站处理器因为是客户端发送来的数据我们要解析。而入站处理器怎么写呢 其实Netty为我们提供了入站出站处理器的多个模板我们需要继承并写上自己的实现就行了。 最简单的入站处理器是SimpleChannelInboundHandler源代码我就不讲了不然又要讲半天。我们新建一个类继承它这个类就叫CustomLengthHandler吧 public class CustomLengthHandler extends SimpleChannelInboundHandlerByteBuf {Overrideprotected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {}
}为什么SimpleChannelInboundHandler的泛型是ByteBuf其实这里不一定是固定的不是第一个处理器的情况你想是什么都可以取决于上一个处理器传递给当前处理器什么东西还记得我们上面的那个流程图吗 一个一个的处理器处理完数据后可以继续往下传递数据传递的数据就是自定义的。例如我从上一个处理器得到ByteBuf对象我将其解析完后封装成一个对象MyObject那么我可以往下传递这个MyObject对象下一个处理器就不用再处理一遍ByteBuf原始数据了下一个处理器直接处理MyBoject封装好数据的对象就行了。类比一下就好像上一个处理器给我当前处理器传递一个JSON字符串我当前处理器处理JSON字符串将其序列化为对象并往下传递这个对象那么下一个处理器就不用再处理原始的JSON字符串了就这么个意思。 所以SimpleChannelInboundHandler的泛型就是上一个处理器传递给当前处理器的数据的类型刚才解释过了它并不是固定的上面的CustomLengthHandler也可以这么写 public class CustomLengthHandler extends SimpleChannelInboundHandlerString {Overrideprotected void channelRead0(ChannelHandlerContext ctx, String str) throws Exception {// 上一个处理器给我传递了一个字符串}
}也可以 public class CustomLengthHandler extends SimpleChannelInboundHandlerInteger {Overrideprotected void channelRead0(ChannelHandlerContext ctx, Integer itg) throws Exception {// 上一个处理器给我传递了一个数字}
}并不是固定的。 好了不说废话了开始代码实现 因为我们是第一个入站处理器上面我们也提到过Netty内部会将数据封装成ByteBuf所以我们从上一个处理器接收到的数据其实是一个ByteBuf对象所以第一个处理器的泛型必需为ByteBuf public class CustomLengthHandler extends SimpleChannelInboundHandlerByteBuf {Overrideprotected void channelRead0(ChannelHandlerContext ctx, ByteBuf buf) throws Exception {}
}ByteBuf是一个字节缓冲区我们可以从它读取到字节数据例如 protected void channelRead0(ChannelHandlerContext ctx, ByteBuf buf) throws Exception {byte b buf.readByte();// 读取1个字节int i buf.readInt();// 读取4个字节因为我们之前说了Java的int是4字节组成的buf.readShort();// 依次类推读取2字节buf.readLong();String str buf.readBytes(5).toString(StandardCharsets.UTF_8);// 读取5个字节并转为字符串注意编码为UTF8
}还记得吗在我们的SP协议中我们定义前四个字节是报文长度所以一开始我们先读取4字节 protected void channelRead0(ChannelHandlerContext ctx, ByteBuf buf) throws Exception {int msgLength buf.readInt();// 报文长度
}在得到这个报文长度字段后我们需要对ByteBuf的长度做一下判断如果它的长度小于报文长度那就说明数据还未全部到达那我们先不做处理等完全到达后再做处理代码像这样 protected void channelRead0(ChannelHandlerContext ctx, ByteBuf buf) throws Exception {int msgLength buf.readInt();if (buf.readableBytes() msgLength){ // 缓冲区中的数据不足 msgLength 个暂不处理return;}// 读取 msgLength 个字节也就是整个报文长度的字节它得到的就是整个报文的完整字节缓冲区ByteBuf bufNew buf.readBytes(msgLength);// 读取 msgLength 个字节不包含 msgLength 占用的4字节// 为了效率也可以写为// ByteBuf bufNew buf.readSlice(msgLength);ctx.fireChannelRead(bufNew);// 传递给下一个处理器
}为什么要这样写还记得一开始我提到的TCP粘包半包吗因为数据并不是一次完整到达的所以我们必需处理数据部分达到的情况。ByteBuf就像一个蓄水池从管道中一开始流进来一些水但是这些水没有达到蓄水池该有的蓄水量所以不管它等它满足了蓄水量我们再处理。 buf.readBytes(msgLength);就是一次性从蓄水池ByteBuf中获取msgLength量的水字节并将它放到一个新的水池ByteBuf bufNew中这个新的水池包含了完整的水量报文所有字节接着往下传递这个新的水池ctx.fireChannelRead(bufNew); 定义完处理器后还需要将它添加进处理器链中还记得我们上面一开始定义的public class CustomHandler extends ChannelInitializerSocketChannel吗在其中添加 public class CustomHandler extends ChannelInitializerSocketChannel {Overrideprotected void initChannel(SocketChannel channel) throws Exception {ChannelPipeline pipeline channel.pipeline();pipeline.addLast(new CustomLengthHandler());// 我们自定义的第一个长度处理器它也是入站处理器1}
}到此为止这个超级简单的报文长度处理器就写完了当然这个处理器有很多的问题它只作为演示实际使用会有很多Bug因为实际使用中要处理的情况有点复杂好在Netty给我们提供了一个开箱即用的报文长度处理器这也是为什么我写得这么简单的原因因为只需了解简单的原理而不需要深入探索Netty有现成的。 这个处理器就是 LengthFieldBasedFrameDecoder它的构造函数常用且重要的有5个参数类型都是int我们一个一个来看 1、第一个参数maxFrameLength是整个报文最大长度说白了就是限制报文大小的你的报文不可能无限大。 2、第二个参数lengthFieldOffset是你的长度字段是从第几个字节开始的我们的SP协议定义了一开始就是长度字段所以这个参数我们可以填0。 3、第三个参数lengthFieldLength是你的长度字段占几个字节我们定义的SP协议指明了长度字段占4个字节所以填4就行。 4、第四个参数lengthAdjustment有点绕是指没有计算进长度但是在报文中存在的数据的长度。例如你有数据5ab因为长度字段5占用4个字节b占用1个字节但是没有把a占用的1个字节算进来所以这个例子中lengthAdjustment就得填1如果是6ab那么lengthAdjustment就得填0因为你将a占用的1字节算进来了。 5、第五个参数initialBytesToStrip是指最终得到的数据要跳过几个字节在我们的SP协议中如果接下来的数据你不想要长度字段那就可以跳过长度字段的4字节initialBytesToStrip就可以填4那么得到的数据中就不包含长度了。 基于我们的SP协议最终得到的处理器应该是 public class CustomHandler extends ChannelInitializerSocketChannel {Overrideprotected void initChannel(SocketChannel channel) throws Exception {ChannelPipeline pipeline channel.pipeline();// 长度处理器它也是入站处理器1pipeline.addLast(new LengthFieldBasedFrameDecoder((1024 * 1024) * 50, // 限制最大报文长度为50MB0, 4, 0, 0));// 长度是从0开始的长度字段4字节偏移量为0不跳过字节}
}②、魔数校验 长度处理完了现在TCP粘包半包所带来的问题我们解决了接下来就是校验魔数新增一个入站处理器 public class CustomMagicNumberHandler extends SimpleChannelInboundHandlerByteBuf {Overrideprotected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {}
}从LengthFieldBasedFrameDecoder中传递过来的数据依旧是ByteBuf所以泛型我们依旧写成ByteBuf到达这里的数据其实还是原始的报文数据只不过经过前面的处理它一定是完整的。 做一下简单的魔数校验 public class CustomMagicNumberHandler extends SimpleChannelInboundHandlerByteBuf {Overrideprotected void channelRead0(ChannelHandlerContext ctx, ByteBuf buf) throws Exception {buf.readInt();// 跳过开头的4字节长度字段int magicNumber buf.readInt();if (magicNumber ! 0xCAFEBABE){ctx.close();// 魔数不正确直接关闭连接}ctx.fireChannelRead(buf);}
}将处理器添加进处理器链 public class CustomHandler extends ChannelInitializerSocketChannel {Overrideprotected void initChannel(SocketChannel channel) throws Exception {ChannelPipeline pipeline channel.pipeline();// 长度处理器它也是入站处理器1pipeline.addLast(new LengthFieldBasedFrameDecoder((1024 * 1024) * 50, // 限制最大报文长度为50MB0, 4,0, 0));// 长度是从0开始的长度字段4字节偏移量为0不跳过字节pipeline.addLast(new CustomMagicNumberHandler());// 魔数处理器入站处理器2}
}③、为客户端生成唯一值UUID或校验客户端的UUID是否存在 这里我就不写了其实就是简单的颁发身份证明和校验身份证明而已生成一个唯一值然后存储到服务器上这里判断UUID是否存在在报文中如果不存在为其生成一个UUID并存储如果存在从服务器存储的UUID中找看能不能找得到。 后面的代码可以根据协议定义的规则解析。 ④、其他规则实现 … Ⅳ、需要注意的点
①、ByteBuf的读取 ByteBuf在读取的时候是不可回退的就像迭代器迭代到下一个就不能再回去读上一个了要想回去重新读必需得重置读取 buf.resetReaderIndex();然后又从最开头开始读取。ByteBuf中数据的基本单位是字节readInt()、readLong()等方法实际上读取的都是字节只不过封装了一下将多个字节转为对应Java类型了。 ②、字符编码 注意解析协议时客户端与服务器都要使用相同的字符编码否则解析字节会对不上因为有些字符编码使用的字节数可能不太一样。 ③、业务逻辑处理 协议解析完后将数据传递到业务逻辑时可以使用Netty服务器启动时的 NioEventLoopGroup worker new NioEventLoopGroup();worker来处理业务逻辑worker的本质其实是一个线程池。 其他的注意事项我想起来了后续会加有什么问题可以评论区留言看到会回复。 五、简单封装的框架 根据以上代码的思路我封装了一个简单的开源框架主要处理SP协议的加强版它包含了长度处理、魔数、客户端标识、路径处理、数据加密等操作暂未做数据验证。 源代码链接是simple-netty-core丢在gitee上了为什么不是GitHub因为我的电脑不科学上网的话始终访问不到GitHub即使修改了host文件也访问不到所以干脆就将源代码丢在gitee上了。 这个框架是我学习Netty时写的比较简单基本能使用感兴趣的可以参考一下也欢迎贡献。 写在最后 最后叠个甲吧以上内容是我个人理解不保证全部正确如有遗漏、错误等后续我会回来更新这篇博客欢迎评论区指正。
