Netty源码分析4-ByteToMessageDecoder

栏目: 编程工具 · 发布时间: 6年前

内容简介:例如应用中使用protobuf协议,则可以将byte转换为Protobuf对象,再交给后面的Handler来处理。下面这段代码, 先将收到的ByteBuf数据按照换行符分割成一段一段的ByteBuf,然后将ByteBuf转换成String, 再将String转换成int, 最后把int加一后写回。

ByteToMessageDecoder 是一种常用的 ChannelInboundHandler ,可以称为解码器,负责将byte字节流(ByteBuf)转换成一种Message,Message是应用可以自己定义的一种 Java 对象。

例如应用中使用protobuf协议,则可以将byte转换为Protobuf对象,再交给后面的Handler来处理。

使用示例

下面这段代码, 先将收到的ByteBuf数据按照换行符分割成一段一段的ByteBuf,然后将ByteBuf转换成String, 再将String转换成int, 最后把int加一后写回。

Netty源码分析4-ByteToMessageDecoder

ServerBootstrap bootstrap = new ServerBootstrap();
EventLoopGroupbossGroup=newNioEventLoopGroup();
EventLoopGroupworkerGroup=newNioEventLoopGroup();
try{
   bootstrap.channel(NioServerSocketChannel.class)
            .handler(newLoggingHandler(LogLevel.DEBUG))
            .group(bossGroup, workerGroup)
            .childHandler(newChannelInitializer<SocketChannel>() {
               protectedvoidinitChannel(SocketChannelch)throwsException{
                   ch.pipeline().addLast(newLineBasedFrameDecoder(1024))
                            .addLast(newByteToStringDecoder())
                            .addLast(newStringToIntegerDecoder())
                            .addLast(newIntegerToByteEncoder())
                            .addLast(newIntegerIncHandler());
                }
            });
   ChannelFuturebind=bootstrap.bind(8092);
   bind.sync();
   bind.channel().closeFuture().sync();
}finally{
   bossGroup.shutdownGracefully().sync();
   workerGroup.shutdownGracefully().sync();
}

这里的ChannelPipeline的组织结构是

1. ByteToStringDecoder - 将byte转换成String的Decoder

2. StringToIntegerDecoder - String转换成Integer对象的Decoder

3. IntegerToByteEncoder - Integer转换成byte的Encoder

4. IntegerIncHandler - 将接受到的int加一后返回

###ByteToStringDecoder

本文主要讲解ByteToMessageDecoder所以先分析ByteToStringDecoder再分析ByteToMessageDecoder

ByteToStringMessageDecoder继承于ByteToMessageDecoder,并实现了ByteToMessageDecoder的

decode(ChannelHandlerContext ctx, ByteBuf in, java.util.List<Object> out) 方法。

decode方法实现中要求将ByteBuf中的数据进行解码然后将解码后的对象增加到list中

public class ByteToStringDecoderextends ByteToMessageDecoder{
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
        byte[] data = new byte[byteBuf.readableBytes()];
        byteBuf.readBytes(data);
        list.add(new String(data, StandardCharsets.UTF_8));
    }
}

ByteToMessageDecoder

ByteToMessageDecoder继承了ChannelInboundHandlerAdapter所以是一个处理Inbound事件的Handler。

其内部保存一个Cumulator用于保存待解码的ByteBuf,然后不断调用子类需要实现的抽象方法decode去取出byte数据转换处理。

/**
     * Cumulate {@link ByteBuf}s.
     */
    public interface Cumulator {
        /**
         * Cumulate the given {@link ByteBuf}s and return the {@link ByteBuf} that holds the cumulated bytes.
         * The implementation is responsible to correctly handle the life-cycle of the given {@link ByteBuf}s and so
         * call {@link ByteBuf#release()} if a {@link ByteBuf} is fully consumed.
         */
        ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in);
    }

Cumulator有两种实现,MERGE_CUMULATOR和COMPOSITE_CMUMULATOR。MERGE_CUMULATOR通过memory copy的方法将in中的数据复制写入到cumulation中。COMPOSITE_CUMULATOR采取的是类似链表的方式,没有进行memory copy, 通过一种CompositeByteBuf来实现,在某些场景下会更适合。默认采用的是MERGE_CUMULATOR。

public static final Cumulator MERGE_CUMULATOR = new Cumulator() {
        @Override
        public ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in) {
            final ByteBuf buffer;
            if (cumulation.writerIndex() > cumulation.maxCapacity() - in.readableBytes()
                    || cumulation.refCnt() > 1 || cumulation.isReadOnly()) {
                // Expand cumulation (by replace it) when either there is not more room in the buffer
                // or if the refCnt is greater then 1 which may happen when the user use slice().retain() or
                // duplicate().retain() or if its read-only.
                // 如果cumulation是只读的、或者要超过capacity了,或者还有其他地方在引用, 则都通过创建一个新的byteBuf的方式来扩容ByteBuf
                buffer = expandCumulation(alloc, cumulation, in.readableBytes());
            } else {
                buffer = cumulation;
            }
            buffer.writeBytes(in);
            in.release();
            return buffer;
        }
    };

ByteToMessageDecoder中最主要的部分在channelRead处理上

  1. 收到一个msg后先判断是否是ByteBuf类型,是的情况创建一个CodecOutputList(也是一种list)保存转码后的对象列表
  2. 如果cumulation为null则把msg设置为cumulation,否则合并到cumulation里
  3. 调用callDecode方法,尝试解码
  4. finally中如果cumulation已经读完了,就release并置为null等待gc
  5. 调用fireChannelRead将解码后的out传递给后面的Handler
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception{
            if (msg instanceof ByteBuf) {
                CodecOutputList out = CodecOutputList.newInstance();
                try {
                    ByteBuf data = (ByteBuf) msg;
                    first = cumulation == null;
                    if (first) {
                        cumulation = data;
                    } else {
                        cumulation = cumulator.cumulate(ctx.alloc(), cumulation, data);
                    }
                    callDecode(ctx, cumulation, out);
                } catch (DecoderException e) {
                    throw e;
                } catch (Exception e) {
                    throw new DecoderException(e);
                } finally {
                    if (cumulation != null && !cumulation.isReadable()) {
                        numReads = 0;
                        cumulation.release();
                        cumulation = null;
                    } else if (++ numReads >= discardAfterReads) {
                        // We did enough reads already try to discard some bytes so we not risk to see a OOME.
                        // See https://github.com/netty/netty/issues/4275
                        numReads = 0;
                        discardSomeReadBytes();
                    }
    
                    int size = out.size();
                    decodeWasNull = !out.insertSinceRecycled();
                    fireChannelRead(ctx, out, size);
                    out.recycle();
                }
            } else {
                ctx.fireChannelRead(msg);
            }
        }
    

callDecode中不断执行抽象decode(ctx, in, out)方法直到in可读数据没有减少或当前handler被remove。

protected void callDecode(ChannelHandlerContext ctx, ByteBufin, List<Object>out){
       try {
           while (in.isReadable()) {
               int outSize = out.size();
               if (outSize > 0) {
                   fireChannelRead(ctx, out, outSize);
                   out.clear();
                  // 检查当前handler是否被remove了
                   if (ctx.isRemoved()) {
                       break;
                   }
                   outSize = 0;
               }

               int oldInputLength = in.readableBytes();
               decodeRemovalReentryProtection(ctx, in, out);
               // 检查当前handler是否被remove了
               if (ctx.isRemoved()) {
                   break;
               }

               if (outSize == out.size()) {
                   if (oldInputLength == in.readableBytes()) {
                       break;
                   } else {
                       continue;
                   }
               }

               if (oldInputLength == in.readableBytes()) { // 这种情况是解码出了对象但是并没有移动in的readIndex
                   throw new DecoderException(
                           StringUtil.simpleClassName(getClass()) +
                                   ".decode() did not read anything but decoded a message.");
               }

               if (isSingleDecode()) {
                   break;
               }
           }
       } ...
   }

fireChannelRead(ctx, msgs, numElements)的处理方式是对每个解码后的消息进行fireChannelRead,交给下一个Handler处理

/**
    static void fireChannelRead(ChannelHandlerContext ctx, List<Object> msgs, int numElements) {
        if (msgs instanceof CodecOutputList) {
            fireChannelRead(ctx, (CodecOutputList) msgs, numElements);
        } else {
            for (int i = 0; i < numElements; i++) {
                ctx.fireChannelRead(msgs.get(i));
            }
        }
    }
    static void fireChannelRead(ChannelHandlerContext ctx, CodecOutputList msgs, int numElements) {
        for (int i = 0; i < numElements; i ++) {
            ctx.fireChannelRead(msgs.getUnsafe(i));
        }
    }

以上就是ByteToMessageDecoder的主要处理部分。关于Netty,面试中会喜欢问道“粘包/拆包”问题,指的是一个消息在网络中是二进制byte流的形式传过去的,接收方如何判断一个消息是否读完、哪里是分割点等,例如构建一个需要的对象需要100byte,但是ByteToMessageDecoder中只取到了50bytes,这时该如何处理。这些可以通过Netty中提供的一些Decoder来实现,例如DelimiterBasedFrameDecoder,FixedLengthFrameDecoder, LengthFieldBasedFrameDecoder。其中最常见的应该是LengthFieldBasedFrameDecoder了,因为自定义的协议中通常会有一个协议头,里面有一个字段描述一个消息的大小长度,然后接收方就能知道消息读到什么时候是读完一个Frame了。这些解码器会在后续的文章中介绍。


以上所述就是小编给大家介绍的《Netty源码分析4-ByteToMessageDecoder》,希望对大家有所帮助,如果大家有任何疑问请给我留言,小编会及时回复大家的。在此也非常感谢大家对 码农网 的支持!

查看所有标签

猜你喜欢:

本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们

React

React

卓越开发者联盟 / 寸志、范洪春、杨森、陈涌 / 电子工业出版社 / 2015-5-1 / CNY 65.00

2014 年横空出世的由Facebook 推出的开源框架React.js,基于Virtual DOM 重新定义了用户界面的开发方式,彻底革新了大家对前端框架的认识,将PHP 风格的开发方式迁移到客户端应用开发。其优势在于可以与各种类库、框架搭配使用。《React:引领未来的用户界面开发框架》是这一领域的首作,由多位一线专家精心撰写,采用一个全程实例全面介绍和剖析了ReactReact.js 的方方......一起来看看 《React》 这本书的介绍吧!

在线进制转换器
在线进制转换器

各进制数互转换器

随机密码生成器
随机密码生成器

多种字符组合密码

正则表达式在线测试
正则表达式在线测试

正则表达式在线测试