郑州网站制作哪家招聘,wordpress好的主题,免费建网站教程,外观设计网站highlight: arduino-light Netty如何实现自定义通信协议 在学习完如何设计协议之后#xff0c;我们又该如何在 Netty 中实现自定义的通信协议呢#xff1f;其实 Netty 作为一个非常优秀的网络通信框架#xff0c;已经为我们提供了非常丰富的编解码抽象基类#xff0c;帮助我… highlight: arduino-light Netty如何实现自定义通信协议 在学习完如何设计协议之后我们又该如何在 Netty 中实现自定义的通信协议呢其实 Netty 作为一个非常优秀的网络通信框架已经为我们提供了非常丰富的编解码抽象基类帮助我们更方便地基于这些抽象基类扩展实现自定义协议。 首先我们看下 Netty 中常用的编解码器有哪些。 一次编解码器和二次编解码器 Netty中的编解码器分为一次编解码和二次编解码。 一次编解码器MessageToByteEncoder、ByteToMessageDecoder/ReplyingDecoder 二次编解码器MessageToMessageEncoder、MessageToMessageDecoder 以解码为例一次解码器用于解决TCP拆包/粘包问题解析得到字节数据。 如果需要对解析后的字节数据做对象转换需要使用二次解码器。同理编码器是相反过程。 为什么需要二次码/编码 假设我们把解决半包粘包问题的常用三种解码器叫一次解码器 那么我们在项目中除了可选的的压缩解压缩之外还需要一层解码因为一次解码的结果是字节需要和项目中所使用的对象做转化方便使用这层解码器可以称为“二次解码器”。 相应的对应的二次编码器是为了将 Java 对象转化成字节流方便存储或传输。 为什么不合并一次二次解码器 思考是不是也可以一步到位 合并 1 次解码(解决粘包、半包)和 2 次解码(解决可操作问题) 可以但是不建议 •没有分层不够清晰;分层可以组合。 •耦合性高不容易置换方案。 常用的编解码方式 -Java 序列化 -Marshaling -XML -JSON -MessagePack -Protobuf -其他 选择编解码方式的因素 -空间编码后占用空间 -时间编解码速度 -是否追求可读性 -是否支持多语言,例如msgpack的支持:Java\C\Python等 Protobuf -Protobuf 是一个灵活的、高效的用于序列化数据的协议。 -相比较 XML 和 JSON 格式Protobuf 更小、更快、更便捷。 -Protobuf 是跨语言的并且自带了一个编译器(protoc)只需要用它进行编译可以自动生成 Java、python、C 等代码不需要再写其他代码。 Protobuf使用步骤 第1步在Maven 项目中引入 Protobuf 坐标下载相关的jar包。 在pom.xml中 添加依赖 xml dependencies dependency groupIdcom.google.protobuf/groupId artifactIdprotobuf-java/artifactId version3.6.1/version /dependency /dependencies 第 2 步 编写proto文件:Student.proto。 Student.proto的内容 java syntax proto3; //版本 option javaouterclassname StudentPoJO; //指定生成的Java类名 //内部类的名称是真正的PoJo 类 message Student{ // message 的规定的 int32 id 1; //PoJo 类的属性数据类型类型和 序号(不是属性值) string name 2; } 第 3 步通过 protoc.exe 根据描述文件生成 Java 类。 说明protoc-3.6.1-win32 是从网上下载的 google 提供的文件. cmd执行命令生成StudentPoJO.java: C:\Users\Administrator\Desktop\Netty资料\我的\资料\protoc-3.6.1-win32\binprotoc.exe --java_out. Student.proto 第4步把生成的 StudentPoJo.java 拷贝到自己的项目中打开。 第 5 步在 Netty 中使用。 java ch.pipeline().addLast(new ProtobufVarint32FrameDecoder()); ch.pipeline().addLast(new ProtobufDecoder(StudentPoJO.Student.getDefaultInstance())); ch.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender()); ch.pipeline().addLast(new ProtobufEncoder()); 抽象编码类 所有的解码器都继承了ChannelInBoundHandler。因为解码是需要解码接收的数据。所以使用In。 所有的编码器都继承了ChannelOutBoundHandler。因为编码是需要将对外发送的数据编码。所以使用Out。 通过抽象编码类的继承图可以看出编码类是 ChanneOutboundHandler 的抽象类实现,具体操作的是 Outbound 出站数据。 常用编码器类型 MessageToByteEncoder 对象编码成字节流MessageToMessageEncoder 一种对象消息类型编码成另外一种对象消息类型。 使用一次编码器IntegerEncoder和二次编码器IntegerToStringEncoder,将消息从Integer编码为String。 java class IntegerEncoder extends MessageToByteEncoderInteger { Override public void encode(ChannelHandlerContext ctx, Integer msg, ByteBuf out) throws Exception { out.writeInt(msg); } } class IntegerToStringEncoder extends MessageToMessageEncoderInteger { Override public void encode(ChannelHandlerContext ctx, Integer message, ListObject out) throws Exception { out.add(message.toString()); } } 使用一次编码器StringEncoder和二次编码器StringToIntegerEncoder,将消息从String编码为Integer。 class StringEncoder extends MessageToByteEncoderString { Override public void encode(ChannelHandlerContext ctx, String msg, ByteBuf out) throws Exception { out.writeCharSequence(msg, Charset.defaultCharset()); } } class StringToIntegerEncoder extends MessageToMessageEncoderString { Override public void encode(ChannelHandlerContext ctx, String message, ListObject out) throws Exception { out.add(Integer.parseInt(message)); } } 编码器MessageToByteEncoder MessageToByteEncoder用于将对象编码成字节流MessageToByteEncoder 提供了唯一的 encode 抽象方法我们需要实现encode 方法即可完成自定义编码。那么encode() 方法是在什么时候被调用的呢 我们一起看下MessageToByteEncoder 的核心源码片段如下所示。 MessageToByteEncoder继承自ChannelOutboundHandlerAdapter。 ChannelOutboundHandlerAdapter 实现了 ChannelOutboundHandler接口,重写了write方法。 这里使用了模板模式:encode方法交给具体的子类实现。 java public abstract class MessageToByteEncoderI extends ChannelOutboundHandlerAdapter { Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { ByteBuf buf null; try { // 1.消息类型是否匹配 不匹配不会处理 // 即传入的是String if (acceptOutboundMessage(msg)) { //I是泛型 SuppressWarnings(unchecked) I cast (I) msg; // 2. 分配 ByteBuf 资源 buf allocateBuffer(ctx, cast, preferDirect); try { // 3. 执行 encode 方法完成数据编码 encode(ctx, cast, buf); } finally { ReferenceCountUtil.release(cast); } if (buf.isReadable()) { // 4. 向后传递写事件 ctx.write(buf, promise); } else { buf.release(); ctx.write(Unpooled.EMPTY_BUFFER, promise); } buf null; } else { ctx.write(msg, promise); } } catch (EncoderException e) { throw e; } catch (Throwable e) { throw new EncoderException(e); } finally { if (buf ! null) { buf.release(); } } } } //供子类重写 protected abstract void encode(ChannelHandlerContext ctx, I msg, ByteBuf out) throws Exception; MessageToByteEncoder 重写了 ChanneOutboundHandler 的 write() 方法其主要逻辑分为以下几个步骤 acceptOutboundMessage 判断是否有匹配的消息类型如果匹配需要执行编码流程如果不匹配直接继续传递给下一个 ChannelOutboundHandler。分配 ByteBuf 资源默认使用堆外内存。调用子类实现的 encode 方法完成数据编码一旦消息被成功编码会通过调用ReferenceCountUtil.release(cast) 自动释放。如果 ByteBuf 可读说明已经成功编码得到数据然后写入 ChannelHandlerContext 交到下一个节点。如果 ByteBuf 不可读则释放 ByteBuf 资源向下传递空的 ByteBuf 对象。 实现类:StringToByteEncoder 编码器实现非常简单不需要关注拆包/粘包问题。 如下例子展示了如何将字符串类型的数据写入到 ByteBuf 实例ByteBuf 实例将传递给 ChannelPipeline 链表中的下一个 ChannelOutboundHandler。 java package io.netty.example.Encode; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.MessageToByteEncoder; public class StringToByteEncoder extends MessageToByteEncoderString { Override protected void encode(ChannelHandlerContext channelHandlerContext, String data, ByteBuf byteBuf) throws Exception { byteBuf.writeBytes(data.getBytes()); } } 编码器MessageToMessageEncoder https://www.javajike.com/book/essential-netty-in-action/chapter4/66cf00f545a4c73fa3fd2fad8d0b7a1d.html MessageToMessageEncoder 与 MessageToByteEncoder 类似同样只需要实现 encode 方法。 与 MessageToByteEncoder 不同的是MessageToMessageEncoder 是将一种格式的消息转换为另外一种格式的消息。 其中第二个 Message 所指的可以是任意一个对象如果该对象是 ByteBuf 类型那么和 MessageToByteEncoder 的实现原理是一致的。 MessageToMessageEncoder继承自ChannelOutboundHandlerAdapter。 ChannelOutboundHandlerAdapter 实现了 ChannelOutboundHandler接口,重写了write方法。 这里使用了模板模式:encode方法交给具体的子类实现。 java Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { CodecOutputList out null; try { // 1. 消息类型是否匹配 不匹配不会处理 // 即传入的是String if (acceptOutboundMessage(msg)) { out CodecOutputList.newInstance(); //I是泛型 SuppressWarnings(unchecked) I cast (I) msg; try { //执行子类encode完成具体编码操作 encode(ctx, cast, out); } finally { ReferenceCountUtil.release(cast); } //如果输出结果是对象列表out是空 if (out.isEmpty()) { out.recycle(); out null; throw new EncoderException( must produce at least one message.); } } else { ctx.write(msg, promise); } } catch (EncoderException e) { throw e; } catch (Throwable t) { throw new EncoderException(t); } finally { if (out ! null) { final int sizeMinusOne out.size() - 1; // sizeMinusOne等于0说明 out长度是1 if (sizeMinusOne 0) { //写出去 ctx.write(out.getUnsafe(0), promise); } else if (sizeMinusOne 0) { //遍历写出去 if (promise ctx.voidPromise()) { writeVoidPromise(ctx, out); } else { writePromiseCombiner(ctx, out, promise); } } //回收 out.recycle(); } } } 此外 MessageToByteEncoder 的输出结果是对象列表out编码后的结果属于中间对象最终仍然会转化成 ByteBuf 进行传输。 MessageToMessageEncoder 常用的实现子类有 StringEncoder、LineEncoder、Base64Encoder 等。以 StringEncoder 为例看下 MessageToMessageEncoder 的用法。 实现类:StringEncoder java Sharable public class StringEncoder extends MessageToMessageEncoderCharSequence { // TODO Use CharsetEncoder instead. private final Charset charset; /** * Creates a new instance with the current system character set. */ public StringEncoder() { this(Charset.defaultCharset()); } /** * Creates a new instance with the specified character set. */ public StringEncoder(Charset charset) { if (charset null) { throw new NullPointerException(charset); } this.charset charset; } Override protected void encode(ChannelHandlerContext ctx, CharSequence msg, ListObject out) throws Exception { if (msg.length() 0) { return; } //编码以后加入out列表 由父类写出即可 out.add(ByteBufUtil.encodeString(ctx.alloc(), CharBuffer.wrap(msg), charset)); } } 思考:现在有1个Java对象要编码为json字符串后转换为byte传输 如何做呢? 1.继承MessageToMessageEncoder 2.重写encode方法 3.将对象序列化为json 4.将json转为ByteBuffer发送:ByteBufUtil.encodeString(ctx.alloc(), CharBuffer.wrap(json), charset)
