分享web开发知识

注册/登录|最近发布|今日推荐

主页 IT知识网页技术软件开发前端开发代码编程运营维护技术分享教程案例
当前位置:首页 > 教程案例

netty解码器详解(小白也能看懂!)

发布时间:2023-09-06 02:20责任编辑:董明明关键词:暂无标签

什么是编解码器?

  首先,我们回顾一下netty的组件设计:Netty的主要组件有Channel、EventLoop、ChannelFuture、ChannelHandler、ChannelPipe等。

ChannelHandler

  ChannelHandler充当了处理入站和出站数据的应用程序逻辑的容器。例如,实现ChannelInboundHandler接口(或ChannelInboundHandlerAdapter),你就可以接收入站事件和数据,这些数据随后会被你的应用程序的业务逻辑处理。当你要给连接的客户端发送响应时,也可以从ChannelInboundHandler冲刷数据。你的业务逻辑通常写在一个或者多个ChannelInboundHandler中。ChannelOutboundHandler原理一样,只不过它是用来处理出站数据的。

ChannelPipeline

  ChannelPipeline提供了ChannelHandler链的容器。以客户端应用程序为例,如果事件的运动方向是从客户端到服务端的,那么我们称这些事件为出站的,即客户端发送给服务端的数据会通过pipeline中的一系列ChannelOutboundHandler,并被这些Handler处理,反之则称为入站的

编码解码器

  当你通过Netty发送或者接受一个消息的时候,就将会发生一次数据转换。入站消息会被解码:从字节转换为另一种格式(比如java对象);如果是出站消息,它会被编码成字节。

  Netty提供了一系列实用的编码解码器,他们都实现了ChannelInboundHadnler或者ChannelOutcoundHandler接口。在这些类中,channelRead方法已经被重写了。以入站为例,对于每个从入站Channel读取的消息,这个方法会被调用。随后,它将调用由已知解码器所提供的decode()方法进行解码,并将已经解码的字节转发给ChannelPipeline中的下一个ChannelInboundHandler。

解码器

抽象基类ByteToMessageDecoder

由于你不可能知道远程节点是否会一次性发送一个完整的信息,tcp有可能出现粘包拆包的问题,这个类会对入站数据进行缓冲,直到它准备好被处理。

主要api有两个:

public abstract class ByteToMessageDecoder extends ChannelInboundHandlerAdapter {
???protected abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception;
???protected void decodeLast(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { ???????if (in.isReadable()) { ???????????// Only call decode() if there is something left in the buffer to decode. ???????????// See https://github.com/netty/netty/issues/4386 ???????????decodeRemovalReentryProtection(ctx, in, out); ???????} ???}}

decode方法:

  必须实现的方法,ByteBuf包含了传入数据,List用来添加解码后的消息。对这个方法的调用将会重复进行,直到确定没有新的元素被添加到该List,或者该ByteBuf中没有更多可读取的字节时为止。然后如果该List不会空,那么它的内容将会被传递给ChannelPipeline中的下一个ChannelInboundHandler。

decodeLast方法:

  当Channel的状态变成非活动时,这个方法将会被调用一次。

最简单的例子:

public class ToIntegerDecoder extends ByteToMessageDecoder { ???@Override ???protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { ???????if (in.readableBytes() >= 4) { ???????????out.add(in.readInt()); ???????} ???}}

这个例子,每次入站从ByteBuf中读取4字节,将其解码为一个int,然后将它添加到下一个List中。当没有更多元素可以被添加到该List中时,它的内容将会被发送给下一个ChannelInboundHandler。int在被添加到List中时,会被自动装箱为Integer。在调用readInt()方法前必须验证所输入的ByteBuf是否具有足够的数据。

一个实用的例子:

public class MyDecoder extends ByteToMessageDecoder { ???@Override ???protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { ???????if (in.readableBytes() < 4) { ???????????return; ???????}
     //在读取前标记readerIndex ???????in.markReaderIndex();
     //读取头部 ???????int length = in.readInt(); ???????if (in.readableBytes() < length) {
     //消息不完整,无法处理,将readerIndex复位 ???????????in.resetReaderIndex(); ???????????return; ???????} ???????out.add(in.readBytes(length).toString(CharsetUtil.UTF_8)); ???}}

抽象类ReplayingDecoder

public abstract class ReplayingDecoder<S> extends ByteToMessageDecoder

  ReplayingDecoder扩展了ByteToMessageDecoder类,使用这个类,我们不必调用readableBytes()方法。参数S指定了用户状态管理的类型,其中Void代表不需要状态管理。

以上代码可以简化为:

public class MySimpleDecoder extends ReplayingDecoder<Void> { ???@Override ???protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { ???????//传入的ByteBuf是ReplayingDecoderByteBuf ???????//首先从入站ByteBuf中读取头部,得到消息体长度length,然后读取length个字节, ????????//并添加到解码消息的List中 ???????out.add(in.readBytes(in.readInt()).toString(CharsetUtil.UTF_8)); ???}

如何实现的?

ReplayingDecoder在调用decode方法时,传入的是一个自定义的ByteBuf实现:

final class ReplayingDecoderByteBuf extends ByteBuf 

ReplayingDecoderByteBuf在读取数据前,会先检查是否有足够的字节可用,以readInt()为例:

final class ReplayingDecoderByteBuf extends ByteBuf { ???private static final Signal REPLAY = ReplayingDecoder.REPLAY; ???...... ????????@Override ???public int readInt() { ???????checkReadableBytes(4); ???????return buffer.readInt(); ???} ???private void checkReadableBytes(int readableBytes) { ???????if (buffer.readableBytes() < readableBytes) { ???????????throw REPLAY; ???????} ???} ?????......}

如果字节数量不够,会抛出一个Error(实际是一个Signal public final class Signal extends Error implements Constant<Signal> ),然后会在上层被捕获并处理,它会把ByteBuf中的ReadIndex恢复到读之前的位置,以供下次读取。当有更多数据可供读取时,该decode()方法将会被再次调用。最终结果和之前一样,从ByteBuf中提取的String将会被添加到List中。

虽然ReplayingDecoder使用方便,但它也有一些局限性:

1. 并不是所有的 ByteBuf 操作都被支持,如果调用了一个不被支持的方法,将会抛出一个 UnsupportedOperationException。

2. ReplayingDecoder 在某些情况下可能稍慢于 ByteToMessageDecoder,例如网络缓慢并且消息格式复杂时,消息被拆成了多个碎片,于是decode()方法会被多次调用反复地解析一个消息。

3. 你需要时刻注意decode()方法在同一个消息上可能被多次调用.。

错误用法:

一个简单的echo服务,客户端在连接建立时,向服务端发送消息(两个1)。服务端需要一次拿到两个Integer,并做处理。

EchoServerHandler

public class EchoServerHandler extends ChannelInboundHandlerAdapter { ???@Override ???public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { ???????System.out.println("msg from client: " + msg); ???} ???@Override ???public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ???} ???@Override ???public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ???????cause.printStackTrace(); ???????ctx.close(); ???}}

EchoClientHandler

public class EchoClientHandler extends ChannelInboundHandlerAdapter { ???@Override ???public void channelActive(ChannelHandlerContext ctx) throws Exception { ???????System.out.println("sent to server: 11"); ???????ctx.writeAndFlush(1); ???????????????Thread.sleep(1000); ???????ctx.writeAndFlush(1); ???} ???@Override ???public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { ???} ???@Override ???public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ???} ???@Override ???public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ???????cause.printStackTrace(); ???????ctx.close(); ???}}

解码器

public class MyReplayingDecoder extends ReplayingDecoder<Void> { ???private final Queue<Integer> values = new LinkedList<>(); ???@Override ???protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { ???????values.add(in.readInt()); ???????values.add(in.readInt()); ???????assert values.size() == 2; ???????out.add(values.poll() + values.poll()); ???}}

运行程序,就会发现断言失败。

我们通过在decode()方法中打印日志或者打断点的方式,可以看到,decode()方法是被调用了两次的,分别在服务端两次接受到消息的时候:

第一次调用时,由于缓冲区中只有四个字节,在第二句 values.add(in.readInt()) 中抛出了异常REPLAY,在ReplayingDecoder中被捕获,并复位ReadIndex。此时values.size() = 1。

第二次调用时,从头开始读取到两个Integer并放入values,因此values.size() = 3。

正确用法:

@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { ???//清空队列 ???values.clear(); ???values.add(in.readInt()); ???values.add(in.readInt()); ???assert values.size() == 2; ???out.add(values.poll() + values.poll());}

如何提高ReplayingDecoder的性能?如上所说,使用ReplayingDecoder存在对一个消息多次重复解码的问题,我们可以通过Netty提供的状态控制来解决这个问题。

首先我们将消息结构设计为:header(4个字节,存放消息体长度),body(消息体)

根据消息的结构,我们定义两个状态:

public enum MyDecoderState { ???/** ????* 未读头部 ????*/ ???READ_LENGTH, ???/** ????* 未读内容 ????*/ ???READ_CONTENT;}

EchoClientHandler

public class EchoClientHandler extends ChannelInboundHandlerAdapter { ???@Override ???public void channelActive(ChannelHandlerContext ctx) throws Exception { ???????for (int i = 0; i < 10; i++) { ???????????System.out.println("sent to server: msg" + i); ???????????ctx.writeAndFlush("msg" + i); ???????} ???} ???@Override ???public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { ???} ???@Override ???public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ???} ???@Override ???public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ???????cause.printStackTrace(); ???????ctx.close(); ???}}

EchoServerHandler

public class EchoServerHandler extends ChannelInboundHandlerAdapter { ???@Override ???public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { ???????System.out.println("msg from client: " + ???????????((ByteBuf) msg).toString(CharsetUtil.UTF_8)); ???} ???@Override ???public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ???} ???@Override ???public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ???????cause.printStackTrace(); ???????ctx.close(); ???}}

解码器

public class IntegerHeaderFrameDecoder extends ReplayingDecoder<MyDecoderState> { ???private int length; ???public IntegerHeaderFrameDecoder() { ???????// Set the initial state. ???????super(MyDecoderState.READ_LENGTH); ???} ???@Override ???protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { ???????switch (state()) { ???????????case READ_LENGTH: ???????????????length = in.readInt(); ???????????????checkpoint(MyDecoderState.READ_CONTENT); ???????????case READ_CONTENT: ???????????????ByteBuf frame = in.readBytes(length); ???????????????checkpoint(MyDecoderState.READ_LENGTH); ???????????????out.add(frame); ???????????????break; ???????????default: ???????????????throw new Error("Shouldn‘t reach here."); ???????} ???}}

编码器

public class MyEncoder extends MessageToByteEncoder<String> { ???@Override ???protected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out) throws Exception { ???????byte[] b = msg.getBytes(); ???????int length = b.length; ???????//write length of msg ???????out.writeInt(length); ???????//write msg ???????out.writeBytes(b); ???}}

当头部被成功读取到时,我们调用 checkpoint(MyDecoderState.READ_CONTENT) 设置状态为“未读消息”,相当于设置一个标志位,如果在后续读取时抛出异常,那么readIndex会被复位到上一次你调用checkpoint()方法的地方。下一次接收到消息,再次调用decode()方法时,就能够从checkpoint处开始读取,避免了又从头开始读。

更多解码器:

LineBasedFrameDecoder

这个类在Netty内部也有使用,它使用行尾控制字符(\n或者\r\n)作为分隔符来解析数据。

DelimiterBasedFrameDecoder

使用自定义的特殊字符作为消息的分隔符。

HttpObjectDecoder

一个HTTP数据的解码器。

这些解码器也非常实用,下次更新关于这些解码器的原理和详细使用。

更多详细内容参见《netty in action》 或者netty源码的英文注释。

netty解码器详解(小白也能看懂!)

原文地址:https://www.cnblogs.com/yuanrw/p/9866356.html

知识推荐

我的编程学习网——分享web前端后端开发技术知识。 垃圾信息处理邮箱 tousu563@163.com 网站地图
icp备案号 闽ICP备2023006418号-8 不良信息举报平台 互联网安全管理备案 Copyright 2023 www.wodecom.cn All Rights Reserved