Netty——编解码器

程序员文章站 2022-07-14 22:17:36

...

什么是编解码器

每个网络应用程序都必须定义如何解析在两个节点之间来回传输的原始字节，以及如何将其和目标应用程序的数据格式做相互转换。这种转换逻辑由编解码器处理，编解码器由编码器和解码器组成，它们每种都可以将字节流从一种格式转换为另一种格式。那么它们的区别是什么呢?

如果将消息看作是对于特定的应用程序具有具体含义的结构化的字节序列——它的数据。

编码器是将消息转换为适合于传输的格式(最有可能的就是字节流);
解码器则是将网络字节流转换回应用程序的消息格式。

因此，编码器操作出站数据，而解码器处理入站数据。

解码器

Netty 所提供的解码器类覆盖了两个不同的用例:

将字节解码为消息——ByteToMessageDecoder 和 ReplayingDecoder;
将一种消息类型解码为另一种——MessageToMessageDecoder。

因为解码器是负责将入站数据从一种格式转换到另一种格式的，所以Netty 的解码器实现了 ChannelInboundHandler 也不会让你感到意外。

什么时候会用到解码器呢?很简单:每当需要为 ChannelPipeline 中的下一个 ChannelInboundHandler 转换入站数据时会用到。此外，得益于 ChannelPipeline 的设计，可以将多个解码器链接在一起，以实现任意复杂的转换逻辑，这也是 Netty 是如何支持代码的模块化以及复用的一个很好的例子。

抽象类ByteToMessageDecoder

将字节解码为消息(或者另一个字节序列)是一项如此常见的任务，以至于 Netty 为它提供了一个抽象的基类:ByteToMessageDecoder。由于你不可能知道远程节点是否会一次性地发送一个完整的消息，所以这个类会对入站数据进行缓冲，直到它准备好处理。
Netty——编解码器

下面举一个如何使用这个类的示例，假设你接收了一个包含简单 int 的字节流，每个 int 都需要被单独处理。在这种情况下，你需要从入站 ByteBuf 中读取每个 int，并将它传递给 ChannelPipeline 中的下一个 ChannelInboundHandler。为了解码这个字节流，你要扩展 ByteToMessageDecoder 类。(需要注意的是，原子类型的 int 在被添加到 List 中时，会被自动装箱为 Integer。)

Netty——编解码器

每次从入站 ByteBuf 中读取 4 字节，将其解码为一个 int，然后将它添加到一个 List 中。当没有更多的元素可以被添加到该 List 中时，它的内容将会被发送给下一个 Channel- InboundHandler。

//扩展了ByteToMessageDecoder，以将字节解码为特定的格式
public class ToIntegerDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        //检查是否至少有4字节可读（1个int的字节长度）
        if (in.readableBytes() >= 4) {
            //从入站ByteBuf中读取一个int，并将其添加到解码消息的List中
            out.add(in.readInt());
        }
    }
}

虽然 ByteToMessageDecoder 使得可以很简单地实现这种模式，但是你可能会发现，在调用 readInt()方法前不得不验证所输入的 ByteBuf 是否具有足够的数据有点繁琐。下面说的 ReplayingDecoder，它是一个特殊的解码器，以少量的开销消除了这个步骤。

编码器中的引用计数：

对于编码器和解码器来说，其过程也是相当的简单:

一旦消息被编码或者解码，它就会被 ReferenceCountUtil.release(message)调用 自动释放。如果你需要保留引用以便稍后使用，那么你可以调用 ReferenceCountUtil.retain(message) 方法。这将会增加该引用计数，从而防止该消息被释放。

抽象类ReplayingDecoder

ReplayingDecoder扩展了ByteToMessageDecoder类，使得我们不必调用readableBytes()方法。它通过使用一个自定义的ByteBuf实现， ReplayingDecoderByteBuf，包装传入的ByteBuf实现了这一点，其将在内部执行该调用。

//完整声明
public abstract class ReplayingDecoder<S> extends ByteToMessageDecoder

//类型参数 S 指定了用于状态管理的类型，其中 Void 代表不需要状态管理。

//扩展ReplayingDecoder<Void>以将字节解码为消息
public class ToIntegerDecoder2 extends ReplayingDecoder<Void> {
    //传入的ByteBuf是ReplayingDecoderByteBuf
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        //从入站ByteBuf中读取一个int，并将其添加到解码消息的List中
        out.add(in.readInt());
    }
}

和之前一样，从ByteBuf中提取的int将会被添加到List中。如果没有足够的字节可用，这个readInt()方法的实现将会抛出一个Error，其将在基类中被捕获并处理。当有更多的数据可供读取时，该decode()方法将会被再次调用。

请注意 ReplayingDecoderByteBuf 的下面这些方面:

并不是所有的 ByteBuf 操作都被支持，如果调用了一个不被支持的方法，将会抛出一个 UnsupportedOperationException;
ReplayingDecoder 稍慢于 ByteToMessageDecoder。

更多解码器：

io.netty.handler.codec.LineBasedFrameDecoder——这个类在 Netty 内部也有使用，它使用了行尾控制字符(\n 或者\r\n)来解析消息数据;
io.netty.handler.codec.http.HttpObjectDecoder——一个 HTTP 数据的解码器。

抽象类MessageToMessageDecoder

使用该抽象基类可以使消息在

//完整声明
public abstract class MessageToMessageDecoder<I> extends ChannelInboundHandlerAdapter

Netty——编解码器

示例：

我们将编写一个 IntegerToStringDecoder 解码器来扩展 MessageToMessageDecoder。它的 decode()方法会把 Integer 参数转换为它的 String 表示：
Netty——编解码器
解码的 S t r i n g 将被添加到传出的 L i s t 中，并转发给下一个 C h a n n e l I n b o u n d H a n d l e r 。

public class IntegerToStringDecoder extends MessageToMessageEncoder<Integer> {
    @Override
    protected void encode(ChannelHandlerContext ctx, Integer msg, List<Object> out) throws Exception {
        //将Integer消息转换为它的String表示，并将其添加到输出的List中
        out.add(String.valueOf(msg));
    }
}

TooLongFrameException类

由于 Netty 是一个异步框架，所以需要在字节可以解码之前在内存中缓冲它们。因此，不能让解码器缓冲大量的数据以至于耗尽可用的内存。为了解除这个常见的顾虑，Netty 提供了 TooLongFrameException 类，其将由解码器在帧超出指定的大小限制时抛出。

为了避免这种情况，你可以设置一个最大字节数的阈值，如果超出该阈值，则会导致抛出一个 TooLongFrameException(随后会被 ChannelHandler.exceptionCaught()方法捕获)。然后，如何处理该异常则完全取决于该解码器的用户。某些协议(如 HTTP)可能允许你返回一个特殊的响应。而在其他的情况下，唯一的选择可能就是关闭对应的连接。

示例：
使用 TooLongFrameException 来通知 ChannelPipeline 中的其他 ChannelHandler 发生了帧大小溢出的。需要注意的是，如果你正在使用一个可变帧大小的协议，那么这种保护措施将是尤为重要的：

public class SafeByteToMessageDecoder extends ByteToMessageDecoder {
    public static final int MAX_FRAME_SIZE = 1024;
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        int readable = in.readableBytes();
        //检查缓冲区是否有超过MAX_FRAME_SIZE个字节
        if (readable > MAX_FRAME_SIZE) {
            //跳过所有的可读字节，抛出TooLongFrameException并通知ChannelHandler
            in.skipBytes(readable);
            throw new TooLongFrameException("Frame too big!");
        }
        //do something
    }
}

编码器

编码器实现了 ChannelOutboundHandler，并将出站数据从一种格式转换为另一种格式，和我们方才学习的解码器的功能正好相反。Netty 提供了一组类，用于帮助你编写具有以下功能的编码器:

将消息编码为字节；
将消息编码为消息；

抽象类MessageToByteEncoder

Netty——编解码器
这个类只有一个方法，而解码器有两个。原因是解码器通常需要在 Channel 关闭之后产生最后一个消息(因此也就有了 decodeLast()方法。这显然不适用于编码器的场景——在连接被关闭之后仍然产生一个消息是毫无意义的。

示例：
Netty——编解码器
上图展示了 ShortToByteEncoder，其接受一个 Short 类型的实例作为消息，将它编码为Short的原子类型值，并将它写入ByteBuf中，其将随后被转发给ChannelPipeline中的下一个 ChannelOutboundHandler。每个传出的 Short 值都将会占用 ByteBuf 中的 2 字节。

public class ShortToByteEncoder extends MessageToByteEncoder<Short> {
    @Override
    protected void encode(ChannelHandlerContext ctx, Short msg, ByteBuf out) throws Exception {
        //将Short写入ByteBuf
        out.writeShort(msg);
    }
}

Netty 提供了一些专门化的 MessageToByteEncoder，你可以基于它们实现自己的编码器。 WebSocket08FrameEncoder类提供了一个很好的实例。你可以在 io.netty.handler. codec.http.websocketx 包中找到它。

抽象类MessageToMessageEncoder

MessageToMessageEncoder 类的 encode()方法提供了将入站数据从一个消息格式解码为另一种。

Netty——编解码器

示例:
使用 IntegerToStringEncoder 扩展了 MessageToMessageEncoder。编码器将每个出站 Integer 的 String 表示添加到了该 List 中。

public class IntegerToStringEncoder extends MessageToMessageEncoder {
    @Override
    protected void encode(ChannelHandlerContext ctx, Object msg, List out) throws Exception {
        out.add(String.valueOf(msg));
    }
}

抽象的编解码器类

虽然我们一直将解码器和编码器作为单独的实体讨论，但是你有时将会发现在同一个类中管理入站和出站数据和消息的转换是很有用的。Netty 的抽象编解码器类正好用于这个目的，因为它们每个都将捆绑一个解码器/编码器对，以处理我们一直在学习的这两种类型的操作。正如同你可能已经猜想到的，这些类同时实现了 ChannelInboundHandler 和 ChannelOutboundHandler 接口。

为什么我们并没有一直优先于单独的解码器和编码器使用这些复合类呢?因为通过尽可能地将这两种功能分开，最大化了代码的可重用性和可扩展性，这是 Netty 设计的一个基本原则。

抽象类ByteToMessageCodec

让我们来研究这样的一个场景:我们需要将字节解码为某种形式的消息，可能是 POJO，随后再次对它进行编码。ByteToMessageCodec 将为我们处理好这一切，因为它结合了 ByteToMessageDecoder 以及它的逆向——MessageToByteEncoder。

任何的请求/响应协议都可以作为使用ByteToMessageCodec的理想选择。例如，在某个 SMTP的实现中，编解码器将读取传入字节，并将它们解码为一个自定义的消息类型，如 SmtpRequest。而在接收端，当一个响应被创建时，将会产生一个SmtpResponse，其将被编码回字节以便进行传输。
Netty——编解码器

抽象类MessageToMessageCodec

通过使用 MessageToMessageCodec，我们可以在一个单个的类中实现该转换的往返过程。MessageToMessageCodec 是一个参数化的类，定义如下:

public abstract class MessageToMessageCodec<INBOUND_IN,OUTBOUND_IN>

Netty——编解码器
decode()方法是将INBOUND_IN类型的消息转换为OUTBOUND_IN类型的消息，而 encode()方法则进行它的逆向操作。将INBOUND_IN类型的消息看作是通过网络发送的类型，而将OUTBOUND_IN类型的消息看作是应用程序所处理的类型，将可能有所裨益。

示例：
WebSocket 协议
下面关于 MessageToMessageCodec 的示例引用了一个新出的 WebSocket 协议，这个协议能实现 Web 浏览器和服务器之间的全双向通信。

我们的WebSocketConvertHandler 在参数化MessageToMessageCodec时将使用INBOUND_IN类型的WebSocketFrame，以及 OUTBOUND_IN类型的MyWebSocketFrame，后者是WebSocketConvertHandler本身的一个静态嵌套类。

public class WebSocketConvertHandler 
        extends MessageToMessageCodec<WebSocketFrame, WebSocketConvertHandler.MyWebSocketFrame> {


    @Override
    protected void encode(ChannelHandlerContext ctx, MyWebSocketFrame msg, List<Object> out) throws Exception {
        //实例化一个指定子类型的WebSocketFrame
        ByteBuf payload = msg.getData().duplicate().retain();
        switch (msg.getType()) {
            case BINARY:
                out.add(new BinaryWebSocketFrame(payload));
                break;
            case TEXT:
                out.add(new TextWebSocketFrame(payload));
                break;
            case CLOSE:
                out.add(new CloseWebSocketFrame(true, 0, payload));
                break;
            case CONTINUATION:
                out.add(new ContinuationWebSocketFrame(payload));
                break;
            case PONG:
                out.add(new PongWebSocketFrame(payload));
                break;
            case PING:
                out.add(new PingWebSocketFrame(payload));
                break;
            default:
                throw new IllegalStateException("Unsupported websocket msg " + msg);
        }
    }

    //将WebSocketFrame解码为MyWebSocketFrame，并设置FrameType
    @Override
    protected void decode(ChannelHandlerContext ctx, WebSocketFrame msg, List<Object> out) throws Exception {
        ByteBuf paload = msg.content().duplicate().retain();
        if (msg instanceof  BinaryWebSocketFrame) {
            out.add(new MyWebSocketFrame(MyWebSocketFrame.FrameType.BINARY, paload));
        } else
        if (msg instanceof  CloseWebSocketFrame) {
            out.add(new MyWebSocketFrame(MyWebSocketFrame.FrameType.CLOSE, paload));
        } else
        if (msg instanceof  PingWebSocketFrame) {
            out.add(new MyWebSocketFrame(MyWebSocketFrame.FrameType.PING, paload));
        } else
        if (msg instanceof  PongWebSocketFrame) {
            out.add(new MyWebSocketFrame(MyWebSocketFrame.FrameType.PONG, paload));
        } else
        if (msg instanceof  TextWebSocketFrame) {
            out.add(new MyWebSocketFrame(MyWebSocketFrame.FrameType.TEXT, paload));
        } else
        if (msg instanceof  ContinuationWebSocketFrame) {
            out.add(new MyWebSocketFrame(MyWebSocketFrame.FrameType.CONTINUATION, paload));
        } else {
            throw new IllegalStateException("Unsupported websocket msg " + msg);
        }
    }

    public static final class MyWebSocketFrame {
        public enum FrameType {
            BINARY,
            CLOSE,
            PING,
            PONG,
            TEXT,
            CONTINUATION
        }
        private final FrameType type;
        private final ByteBuf data;

        public MyWebSocketFrame(FrameType type, ByteBuf data) {
            this.type = type;
            this.data = data;
        }

        public FrameType getType() {
            return type;
        }

        public ByteBuf getData() {
            return data;
        }
    }
}

CombinedChannelDuplexHandler类

正如我们前面所提到的，结合一个解码器和编码器可能会对可重用性造成影响。但是，有一种方法既能够避免这种惩罚，又不会牺牲将一个解码器和一个编码器作为一个单独的单元部署所带来的便利性。

CombinedChannelDuplexHandler 提供了这个解决方案，其声明为:

public class CombinedChannelDuplexHandler
	<I extends ChannelInboundHandler, O extends ChannelOutboundHandler>

这个类充当了 ChannelInboundHandler 和 ChannelOutboundHandler(该类的类型参数 I 和 O)的容器。通过提供分别继承了解码器类和编码器类的类型，我们可以实现一个编解码器，而又不必直接扩展抽象的编解码器类。

示例：

public class ByteToCharDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        while (in.readableBytes() >= 2) {
            out.add(in.readChar());
        }
    }
}

这里的 decode()方法一次将从 ByteBuf 中提取 2 字节，并将它们作为 char 写入到 List 中，其将会被自动装箱为 Character 对象。

public class CharToByteEncoder extends MessageToByteEncoder<Character> {
    @Override
    protected void encode(ChannelHandlerContext ctx, Character msg, ByteBuf out) throws Exception {
        out.writeChar(msg);
    }
}

将 Character 转换回字节。这个类扩展了 MessageToByteEncoder，因为它需要将 char 消息编码到 ByteBuf 中。这是通过直接写入 ByteBuf 做到的。

//通过该解码器和编码器实现参数化CombinedByteCharCodec
public class CombinedChannelDuplexHandler extends
        io.netty.channel.CombinedChannelDuplexHandler<ByteToCharDecoder, CharToByteEncoder> {
    public CombinedChannelDuplexHandler() {
        //将委托实例传递给父类
        super(new ByteToCharDecoder(), new CharToByteEncoder());
    }
}

Netty——编解码器

什么是编解码器

解码器

抽象类ByteToMessageDecoder

抽象类ReplayingDecoder

抽象类MessageToMessageDecoder

TooLongFrameException类

编码器

抽象类MessageToByteEncoder

抽象类MessageToMessageEncoder

抽象的编解码器类

抽象类ByteToMessageCodec

抽象类MessageToMessageCodec

CombinedChannelDuplexHandler类

Netty解决粘包和拆包问题的四种方案

Netty源码分析（十）----- 拆包器之LineBasedFrameDecoder

史诗级最强教科书式“NIO与Netty编程”

netty之NioEventLoopGroup源码分析二

Netty框架问题记录1--多线程下批量发送消息导致消息被覆盖

NETTY学习经验

Netty 在 Dubbo 中是如何应用的？

netty服务端启动--ServerBootstrap源码解析

Netty中的ChannelFuture和ChannelPromise

Netty：初识Netty

Netty——编解码器

什么是编解码器

解码器

抽象类ByteToMessageDecoder

抽象类ReplayingDecoder

抽象类MessageToMessageDecoder

TooLongFrameException类

编码器

抽象类MessageToByteEncoder

抽象类MessageToMessageEncoder

抽象的编解码器类

抽象类ByteToMessageCodec

抽象类MessageToMessageCodec

CombinedChannelDuplexHandler类

Netty解决粘包和拆包问题的四种方案

Netty源码分析 （十）----- 拆包器之LineBasedFrameDecoder

史诗级最强教科书式“NIO与Netty编程”

netty之NioEventLoopGroup源码分析二

Netty框架问题记录1--多线程下批量发送消息导致消息被覆盖

NETTY学习经验

Netty 在 Dubbo 中是如何应用的？

netty服务端启动--ServerBootstrap源码解析

Netty中的ChannelFuture和ChannelPromise

Netty：初识Netty

Netty源码分析（十）----- 拆包器之LineBasedFrameDecoder