Netty4之编解码

冷不防 2023-10-11 09:29 5阅读 0赞

本文是基于Netty4.1.x，一般在使用Netty作为网络框架进行开发时，编解码框架是我们应该注意的一个重要部分。应用从网络层接收数据需要经过解码(Decode),将二进制的数据报转换从应用层的协议消息，这样才能被应用逻辑所识别。同样，客户端发送或服务端在返回消息时，是需要将消息编码(Encode)成二进制字节数组(在Netty4中就是ByteBuf)对能发送到网络对端。对于编解码Netty4本身提供了一些基本的编解码器，当然如果需要定义自己的私有协议时，需要开发者基于已有的编解码框架来实现自定义的编解码器。下面主要介绍Netty4中的编解码的实现原理及如何自定义一个编解码器。

public class NettyServer {
        public static void main(String[] args) throws Exception {
            new NettyServer().start("127.0.0.1", 8081);
        }
    
        public void start(String host, int port) throws Exception {
            ExecutorService executorService = Executors.newCachedThreadPool();
            EventLoopGroup bossGroup = new NioEventLoopGroup(0, executorService);//Boss I/O线程池,用于处理客户端连接,连接建立之后交给work I/O处理
            EventLoopGroup workerGroup = new NioEventLoopGroup(0, executorService);//Work I/O线程池
            EventExecutorGroup businessGroup = new DefaultEventExecutorGroup(2);//业务线程池
            ServerBootstrap server = new ServerBootstrap();//启动类
            server.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024).childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline pipeline = ch.pipeline();
                    pipeline.addLast("decoder", new StringDecoder());
                    pipeline.addLast("encoder", new StringEncoder());
                    pipeline.addLast(businessGroup, new ServerHandler());
                }
            });
            server.childOption(ChannelOption.TCP_NODELAY, true);
            server.childOption(ChannelOption.SO_RCVBUF, 32 * 1024);
            server.childOption(ChannelOption.SO_SNDBUF, 32 * 1024);
            InetSocketAddress addr = new InetSocketAddress(host, port);
            server.bind(addr).sync().channel();//启动服务
        }
    }

上面例子中在initChannel中分别放置了一个decoder跟一个encoder，其中StringDecoder/StringEncoder是Netty4编解码框架中的一成员。下面看看StringDecoder怎么实现的：

@Sharable
    public class StringDecoder extends MessageToMessageDecoder<ByteBuf> {
    
        // TODO Use CharsetDecoder instead.
        private final Charset charset;
    
        /**
         * Creates a new instance with the current system character set.
         */
        public StringDecoder() {
            this(Charset.defaultCharset());
        }
    
        /**
         * Creates a new instance with the specified character set.
         */
        public StringDecoder(Charset charset) {
            if (charset == null) {
                throw new NullPointerException("charset");
            }
            this.charset = charset;
        }
    
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception {
            out.add(msg.toString(charset));
        }
    }

StringDecoder本身很简单，执行解码(decode)方法时候将二进制字节数组转成了String,其中StringDecoder继承了MessageToMessageDecoder，下面看看MessageToMessageDecoder是怎么做的。

public abstract class MessageToMessageDecoder<I> extends ChannelInboundHandlerAdapter {
    
        private final TypeParameterMatcher matcher;
    
        /**
         * Create a new instance which will try to detect the types to match out of the type parameter of the class.
         */
        protected MessageToMessageDecoder() {
            matcher = TypeParameterMatcher.find(this, MessageToMessageDecoder.class, "I");
        }
    
        /**
         * Create a new instance
         *
         * @param inboundMessageType    The type of messages to match and so decode
         */
        protected MessageToMessageDecoder(Class<? extends I> inboundMessageType) {
            matcher = TypeParameterMatcher.get(inboundMessageType);
        }
    
        /**
         * Returns {@code true} if the given message should be handled. If {@code false} it will be passed to the next
         * {@link ChannelInboundHandler} in the {@link ChannelPipeline}.
         */
        public boolean acceptInboundMessage(Object msg) throws Exception {
            return matcher.match(msg);
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            CodecOutputList out = CodecOutputList.newInstance();
            try {
                if (acceptInboundMessage(msg)) {
                    @SuppressWarnings("unchecked")
                    I cast = (I) msg;
                    try {
                        decode(ctx, cast, out);
                    } finally {
                        ReferenceCountUtil.release(cast);
                    }
                } else {
                    out.add(msg);
                }
            } catch (DecoderException e) {
                throw e;
            } catch (Exception e) {
                throw new DecoderException(e);
            } finally {
                int size = out.size();
                for (int i = 0; i < size; i ++) {
                    ctx.fireChannelRead(out.getUnsafe(i));
                }
                out.recycle();
            }
        }
    
        /**
         * Decode from one message to an other. This method will be called for each written message that can be handled
         * by this decoder.
         *
         * @param ctx           the {@link ChannelHandlerContext} which this {@link MessageToMessageDecoder} belongs to
         * @param msg           the message to decode to an other one
         * @param out           the {@link List} to which decoded messages should be added
         * @throws Exception    is thrown if an error occurs
         */
        protected abstract void decode(ChannelHandlerContext ctx, I msg, List<Object> out) throws Exception;
    }

从上面看出，MessageToMessageDecoder继承了ChannelInboundHandlerAdapter本身就是一个处理器Handler,只是定义了一个decode()方法，当上行消息到达时，channelRead()方法被调用，下面重点来了，看channelRead()方法都执行了那些逻辑。acceptInboundMessage(msg);这个方法表示先判断一下是否应该处理这个消息，如果不处理直接添加到解码的结果中；如果需要处理则执行decode()方法，这里的decode方法具体实现是在其子类也就是StringDecoder中实现的。当消息被解码之后，需要将原消息对象进行释放。最后调用 ctx.fireChannelRead(out.getUnsafe(i));这个方法的含义是将消息传递给下一个InBoundHandler处理器去处理，到这里解码就已经完成了。下面看看编码器StringEncoder:

@Sharable
    public class StringEncoder extends MessageToMessageEncoder<CharSequence> {
    
        // 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, List<Object> out) throws Exception {
            if (msg.length() == 0) {
                return;
            }
    
            out.add(ByteBufUtil.encodeString(ctx.alloc(), CharBuffer.wrap(msg), charset));
        }
    }

编码器本身也很简单，主要是实现了其父类MessageToMessageEncoder的encode方法，将String 编码成ByteBuf。下面看看MessageToMessageEncoder的逻辑：

public abstract class MessageToMessageEncoder<I> extends ChannelOutboundHandlerAdapter {
    
        private final TypeParameterMatcher matcher;
    
        /**
         * Create a new instance which will try to detect the types to match out of the type parameter of the class.
         */
        protected MessageToMessageEncoder() {
            matcher = TypeParameterMatcher.find(this, MessageToMessageEncoder.class, "I");
        }
    
        /**
         * Create a new instance
         *
         * @param outboundMessageType   The type of messages to match and so encode
         */
        protected MessageToMessageEncoder(Class<? extends I> outboundMessageType) {
            matcher = TypeParameterMatcher.get(outboundMessageType);
        }
    
        /**
         * Returns {@code true} if the given message should be handled. If {@code false} it will be passed to the next
         * {@link ChannelOutboundHandler} in the {@link ChannelPipeline}.
         */
        public boolean acceptOutboundMessage(Object msg) throws Exception {
            return matcher.match(msg);
        }
    
        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            CodecOutputList out = null;
            try {
                if (acceptOutboundMessage(msg)) {
                    out = CodecOutputList.newInstance();
                    @SuppressWarnings("unchecked")
                    I cast = (I) msg;
                    try {
                        encode(ctx, cast, out);
                    } finally {
                        ReferenceCountUtil.release(cast);
                    }
    
                    if (out.isEmpty()) {
                        out.recycle();
                        out = null;
    
                        throw new EncoderException(
                                StringUtil.simpleClassName(this) + " 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;
                    if (sizeMinusOne == 0) {
                        ctx.write(out.getUnsafe(0), promise);
                    } else if (sizeMinusOne > 0) {
                        // Check if we can use a voidPromise for our extra writes to reduce GC-Pressure
                        // See https://github.com/netty/netty/issues/2525
                        if (promise == ctx.voidPromise()) {
                            writeVoidPromise(ctx, out);
                        } else {
                            writePromiseCombiner(ctx, out, promise);
                        }
                    }
                    out.recycle();
                }
            }
        }
    
        private static void writeVoidPromise(ChannelHandlerContext ctx, CodecOutputList out) {
            final ChannelPromise voidPromise = ctx.voidPromise();
            for (int i = 0; i < out.size(); i++) {
                ctx.write(out.getUnsafe(i), voidPromise);
            }
        }
    
        private static void writePromiseCombiner(ChannelHandlerContext ctx, CodecOutputList out, ChannelPromise promise) {
            final PromiseCombiner combiner = new PromiseCombiner(ctx.executor());
            for (int i = 0; i < out.size(); i++) {
                combiner.add(ctx.write(out.getUnsafe(i)));
            }
            combiner.finish(promise);
        }
    
        /**
         * Encode from one message to an other. This method will be called for each written message that can be handled
         * by this encoder.
         *
         * @param ctx           the {@link ChannelHandlerContext} which this {@link MessageToMessageEncoder} belongs to
         * @param msg           the message to encode to an other one
         * @param out           the {@link List} into which the encoded msg should be added
         *                      needs to do some kind of aggregation
         * @throws Exception    is thrown if an error occurs
         */
        protected abstract void encode(ChannelHandlerContext ctx, I msg, List<Object> out) throws Exception;

从上面看出，MessageToMessageEncoder继承了ChannelOutboundHandlerAdapter本身就是一个处理器Handler,只是定义了一个encode()方法，当上行消息到达时，wirte()方法被调用，下面重点来了，看write()方法都执行了那些逻辑。(acceptOutboundMessage(msg));这个方法表示先判断一下是否应该处理这个消息，如果不处理直接ctx.write(msg, promise)这个方法表示将结果传递到下一下outBound Handler中；如果需要处理则执行encode()方法，这里的encode方法具体实现是在其子类也就是StringEncoder中实现的。当消息被编码之后，需要将原消息对象进行释放。最后调用 ctx.write(out.getUnsafe(0), promise)或 writeVoidPromise(ctx, out);这个方法的含义是将消息传递给下一个处理器去处理，到这里解码就已经完成了。自此Netty4编解码流程已经完了。总结一下:

*  编解码器本质上就是一个处理器InBound或OutBound Handler,被放置在处理链中；
 *  编解码器核心部分是其自定义的decode或encode方法，具体编解码逻辑都放在这两个方法中进行；
 *  当编解码处理完成之后需要写到下一个处理器Handler,一般在channelRead或write中调用ctx.fireChannelRead()或ctx.write()方法。注意在自定义编解码器时在ChannelRead或write方法中释放相关的流，否则会造成内存泄漏；
 *  一般编解码处理器Hander位于处理链中的最前面，当消息上行时，解码器最先处理;当消息下行时，编码器最后处理

下面看看如何自定义一个编解码器，此处直接实现一个String的编解码器，继承ByteToMessageCodec

public class StringCodec extends ByteToMessageCodec<String> {
    
        //执行编码逻辑
        @Override
        protected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out) throws Exception {
            out.writeBytes(msg.getBytes(StandardCharsets.UTF_8));
    
        }
         //执行解码逻辑
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
            out.add(in.toString(StandardCharsets.UTF_8));
        }
    }

从上面一段代码可以看出自定义一个编码器很简单。Netty4框架为开发者提供了很多默认的编解码器，处理器。让开发者能更加关注自已业务逻辑的开发。