暂无图片
暂无图片
暂无图片
暂无图片
暂无图片
首页 / netty粘包拆包之DelimiterBasedFrameDecoder解码器

netty粘包拆包之DelimiterBasedFrameDecoder解码器

文一西路代码狗 2020-10-25
870
目录
          一、 背景简介
          二、 应用
          三、源码

一、背景简介

DelimiterBasedFrameDecoder 基于自定义分隔符解码器，他能够按照我们自定义的特殊符号或者字符作为分隔符对接收到的消息进行分段解码，在服务端接收到信息进行解析的时候DelimiterBasedFrameDecoder的构造方法需要传递两个参数maxFrameLength和可变长参数delimiters，maxFrameLength用来限制对接收到的数据进行解码的时候一次解码的最大帧长度。如果在解码的过程中对传递的数据进行解码后大于maxLength，并且此时开启了快速失败机制，则会立即抛出异常信息。反之，如果在一次解码的过程中对传递的数据进行解码后小于于maxLength，那就正常的读取解码后的数据信息。可变长参数delimiters表示服务端解码的时候自己定义的分隔符(一个或者多个)，如果只有一个那么就会按照这个分隔符对接收到的数据进行分割，如果为多个，那就会按照自定义的多个分隔符进行解码数据，如果最后一个分隔符后面仍然有数据(即不是以分隔符结尾)，那么最后一个分隔符后面的数据皆不会被解码，实际上还是在缓冲区中，只是并未被读取出来。

二、应用
下面是DelimiterBasedFrameDecoder的一个简单小应用测试，模拟使用不同的分隔符发送不同信息服务端使用DelimiterBasedFrameDecoder解码出来的信息是什么样的。
服务端：
public class DelimiterBasedFrameDecoderTestServer {


    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) {
                            ByteBuf delimiter1 = Unpooled.copiedBuffer("%".getBytes());
                            ByteBuf delimiter2 = Unpooled.copiedBuffer("$".getBytes());
                            ch.pipeline()
                                    .addLast(new DelimiterBasedFrameDecoder(1024, delimiter1, delimiter2))
                                    .addLast(new ChannelInboundHandlerAdapter() {
                                        @Override
                                        public void channelRead(ChannelHandlerContext ctx, Object msg) {
                                            if (msg instanceof ByteBuf) {
                                                ByteBuf packet = (ByteBuf) msg;
                                                System.out.println(packet.toString(Charset.defaultCharset()));
                                            }
                                        }
                                    });
                        }
                    });
            ChannelFuture f = b.bind(9000).sync();
            System.out.println("Started DelimiterBasedFrameDecoderTestServer...");
            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
}
客户端：
public class DelimiterBasedFrameDecoderTestClient {


    public static void main(String[] args) throws Exception {
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(workerGroup)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) {
                            ch.pipeline()
                                    .addLast(new ChannelInboundHandlerAdapter() {
                                        public void channelActive(ChannelHandlerContext ctx) {
                                            ByteBuf byteBuf = Unpooled.copiedBuffer("he111%llo%world$w$orl$$dworld%ttt".getBytes());
                                            ctx.writeAndFlush(byteBuf);
                                        }
                                    });
                        }
                    });
            ChannelFuture f = b.connect("127.0.0.1", 9000).sync();
            System.out.println("Started DelimiterBasedFrameDecoderTestClient...");
            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
        }
    }
}
①、发送的消息以服务端分隔符结尾

②、发送的消息未以服务端分隔符结尾

这里演示的是多个分隔符的情况，单个分隔符类似。
可以看到如果发送的消息中最后一个分隔符后面仍然有数据(即不是以分隔符结尾)，那么最后一个分隔符后面的数据皆不会被解码，实际上还是在缓冲区中，只是并未被读取出来。
三、源码

netty提供的各种解码器统一都继承了ByteToMessageDecoder，ByteToMessageDecoder负责读取字节流并转换成其他消息，而ByteToMessageDecoder又继承了ChannelInboundHandlerAdapter，而ChannelInboundHandlerAdapter正是我们刚才在测试数据的时候在客户端重写了他的方法channelActive(),使管道生效并且数据写入缓冲区并发送数据。

在客户端代码流程原理并不复杂，我们主要看服务端是怎么解码、获取到数据的。
当客户端通过Channel发送数据的时候，服务端会在ByteToMessageDecoder的channelRead方法接收到。我们以channelRead方法为入口看服务端是怎么进行消息解码的。
在channelRead方法中首先会判断消息msg是不是ByteBuf类型的：
①、如果msg不是ByteBuf类型的，直接调用ctx.fireChannelRead(msg)方法，fireChannelRead方法实际上是调用pipeline管道的下一个Handler去继续处理消息其他的逻辑去了，我们进入到AbstractChannelHandlerContext的fireChannelRead方法，发现只有一个方法invokeChannelRead，寻找下一个绑定的Handler并且调用ChannelRead方法。
    @Override
    public ChannelHandlerContext fireChannelRead(final Object msg) {
        //寻找下一个绑定的Handler并且调用ChannelRead方法
        invokeChannelRead(findContextInbound(), msg);
        return this;
    }
继续进入到invokeChannelRead。
    static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
        final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            next.invokeChannelRead(m);
        } else {
            executor.execute(() -> next.invokeChannelRead(m));
        }
    }
最终都会调用invokeChannelRead，而invokeChannelRead就是将Handler转换成ChannelInboundHandler类型去执行具体的channelRead方法，去进一步的读取消息并处理相关Handler逻辑。
    private void invokeChannelRead(Object msg) {
        if (invokeHandler()) {
            try {
                ((ChannelInboundHandler) handler()).channelRead(this, msg);
            } catch (Throwable t) {
                notifyHandlerException(t);
            }
        } else {
            fireChannelRead(msg);
        }
    }
②、如果msg是ByteBuf类型的，去调用解码逻辑callDecode方法去解析数据。
    protected void callDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        try {
            while (in.isReadable()) {
                int outSize = out.size();
                //将out里的事件执行并且清空out
                if (outSize > 0) {
                    fireChannelRead(ctx, out, outSize);
                    out.clear();
                    // 在继续解码之前，请检查是否已删除此处理程序
                    // 如果已将其删除，则继续在缓冲区上操作是不安全的。
                    if (ctx.isRemoved()) {
                        break;
                    }
                    outSize = 0;
                }
                int oldInputLength = in.readableBytes();
                // 开始解析数据，如果解析出来数据，那么out的长度一定会改变
                decode(ctx, in, out);


                // 在继续循环之前，请检查是否已删除此处理程序。
                // 如果已将其删除，则继续在缓冲区上操作是不安全的。
                if (ctx.isRemoved()) {
                    break;
                }
                // 如果没有解析出来数据
                if (outSize == out.size()) {
                    if (oldInputLength == in.readableBytes()) {
                        break;
                    } else {
                        continue;
                    }
                }


                if (oldInputLength == in.readableBytes()) {
                    throw new DecoderException(
                            StringUtil.simpleClassName(getClass()) +
                            ".decode() did not read anything but decoded a message.");
                }
                if (isSingleDecode()) {
                    break;
                }
            }
        } catch (DecoderException e) {
            throw e;
        } catch (Throwable cause) {
            throw new DecoderException(cause);
        }
    }
在callDecode方法中，使用while循环去解析数据，这是因为一次发送的数据有可能被解析成多个分段，对于循环中的每次解析，首先判断缓冲区中是否还有可读空间，有可读空间执行while循环体，第一次执行循环体的时候out肯定是没有数据的，也就是out的size为0，下面会调用解码逻辑decode(ctx, in, out)去进行数据的第一次解析，我们先不看decode(ctx, in, out)到底执行了什么逻辑，现在假设已经执行了一次decode(ctx, in, out)逻辑，也就是out里面有数据了，就是第一次解析出来的数据，现在out的size为1，现在进入outSize > 0逻辑判断中，可以看到这个if判断逻辑核心就是fireChannelRead(ctx, out, outSize)，然后将out清空，size置位0，关于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));
            }
        }
    }
继续进入到fireChannelRead，来到AbstractChannelHandlerContext的fireChannelRead。
    @Override
    public ChannelHandlerContext fireChannelRead(final Object msg) {
        //寻找下一个绑定的Handler并且调用ChannelRead方法
        invokeChannelRead(findContextInbound(), msg);
        return this;
    }
继续往里走。
    static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
        final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            next.invokeChannelRead(m);
        } else {
            executor.execute(() -> next.invokeChannelRead(m));
        }
    }


    private void invokeChannelRead(Object msg) {
        if (invokeHandler()) {
            try {
                ((ChannelInboundHandler) handler()).channelRead(this, msg);
            } catch (Throwable t) {
                notifyHandlerException(t);
            }
        } else {
            fireChannelRead(msg);
        }
    }
重点看((ChannelInboundHandler) handler()).channelRead(this, msg)，这里将handler转成ChannelInboundHandler类型，执行他的channelRead方法，其实就是我们刚才服务端的重写的channelRead。

也就是说解析完一次数据时候立刻就去执行我们自定义的处理逻辑去了。当然在这里只是一个测试代码，如果真正我们自定义的处理逻辑比较复杂，比如涉及到操作数据库、磁盘IO等，可以考虑放入线程池中进行业务处理，避免造成解析数据缓慢，网络迟钝。
回到我们刚才while循环解析数据的逻辑，进入到decode(ctx, in, out)中。进入到其实现类DelimiterBasedFrameDecoder中。
这里我们先看一下DelimiterBasedFrameDecoder中核心属性。
    // ====================核心属性====================
    //自定义的分隔符数量，可以为1个或者多个
    private final ByteBuf[] delimiters;
    //每帧消息的最大长度
    private final int maxFrameLength;
    //解码消息时，是否丢弃分隔符
    private final boolean stripDelimiter;
    //是否开启快速失败机制(遇到错误时，是否立即抛出异常)
    private final boolean failFast;
    //是否正在丢弃一个帧的消息
    private boolean discardingTooLongFrame;
    //丢弃消息的总长度
    private int tooLongFrameLength;
    /** 仅在使用“ \ n”和“ \ r \ n”作为分隔符进行解码时设置. */
    private final LineBasedFrameDecoder lineBasedDecoder;
核心属性都已标明注释，具体作用等下分析源码的时候再详细介绍。
接着进入到decode方法中去。
    @Override
    protected final void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        Object decoded = decode(ctx, in);
        if (decoded != null) {
            out.add(decoded);
        }
    }
decoded就是解析完的对象，如果这个对象不为空，就放到out中去，再看decode方法。
    protected Object decode(ChannelHandlerContext ctx, ByteBuf buffer) throws Exception {
        //如果lineBasedDecoder不为空则用lineBasedDecoder解码器解析
        if (lineBasedDecoder != null) {
            return lineBasedDecoder.decode(ctx, buffer);
        }
        // 遍历所有定界符，然后选择产生最短帧的定界符。
        int minFrameLength = Integer.MAX_VALUE;
        ByteBuf minDelim = null;
        for (ByteBuf delim: delimiters) {
            //依次遍历每一个分隔符，找到产生的最短帧的长度和该分隔符delim
            int frameLength = indexOf(buffer, delim);
            if (frameLength >= 0 && frameLength < minFrameLength) {
                minFrameLength = frameLength;
                minDelim = delim;
            }
        }
        //如果找到了分隔符delim
        if (minDelim != null) {
            int minDelimLength = minDelim.capacity();
            ByteBuf frame;
            //如果找到分隔符delim 并且之前在丢弃模式中
            if (discardingTooLongFrame) {
                // 继续丢弃消息.
                // 回到初始状态.
                discardingTooLongFrame = false;
                buffer.skipBytes(minFrameLength + minDelimLength);


                int tooLongFrameLength = this.tooLongFrameLength;
                this.tooLongFrameLength = 0;
                if (!failFast) {
                    fail(tooLongFrameLength);
                }
                return null;
            }
            //如果找到分隔符delim 并且之前不在丢弃模式中 判断是否超过每帧最大长度
            if (minFrameLength > maxFrameLength) {
                // Discard read frame.
                buffer.skipBytes(minFrameLength + minDelimLength);
                fail(minFrameLength);
                return null;
            }
            //如果找到分隔符delim 并且之前不在丢弃模式中 没有超过每帧最大长度 那么就是正常读取消息 正常的流程
            if (stripDelimiter) {
                frame = buffer.readRetainedSlice(minFrameLength);
                buffer.skipBytes(minDelimLength);
            } else {
                frame = buffer.readRetainedSlice(minFrameLength + minDelimLength);
            }


            return frame;
            //如果没有找到分隔符delim
        } else {
            //如果没有找到分隔符delim 判断之前是否在丢弃模式中
            if (!discardingTooLongFrame) {
                //如果没有找到分隔符delim 并且之前不在丢弃模式中 那么就判断是否超过每帧最大长度
                if (buffer.readableBytes() > maxFrameLength) {
                    // 丢弃缓冲区的内容，直到找到定界符.
                    tooLongFrameLength = buffer.readableBytes();
                    buffer.skipBytes(buffer.readableBytes());
                    discardingTooLongFrame = true;
                    // 抛出异常
                    if (failFast) {
                        fail(tooLongFrameLength);
                    }
                }
            } else {
                //如果没有找到分隔符delim 并且之前在丢弃模式中
                // 由于找不到分隔符，因此仍在丢弃缓冲区.
                tooLongFrameLength += buffer.readableBytes();
                buffer.skipBytes(buffer.readableBytes());
            }
            return null;
        }
    }
这个方法的处理逻辑就是DelimiterBasedFrameDecoder解码器的核心解码逻辑。
首先判断lineBasedDecoder 是否为空，如果lineBasedDecoder不为空则用lineBasedDecoder解码器解析，就是以换行符的解码方式对数据进行解码。然后遍历所有定界符，然后选择从数据头开始产生最短帧的定界符。依次遍历每一个分隔符，找到产生的最短帧的长度和该分隔符delim，并记录下来。
接下来如果找到了这样的一个分隔符，然后discardingTooLongFrame判断之前是否在丢弃模式中，如果找到分隔符delim 并且之前在丢弃模式中，那么继续丢弃消息.。反之，如果找到分隔符delim 并且之前不在丢弃模式中 判断是否超过每帧最大长度，如果超过了最大长度，那么久跳过这个消息，并且开启了快速失败的话，那么久立即抛出异常。反之，如果没有超过最大长度，那么就是正常的解码逻辑，正常读取数据就行了。
如果没有找到这样的一个分隔符，判断之前是否在丢弃模式中，如果之前不在丢弃模式中 那么就判断是否超过每帧最大长度，如果超过了最大长度，那么就跳过这个数据并且抛出异常，如果么有超过最大长度，那么就返回null，因为下次发动的数据可能就满足条件了，就能正常读取了。如果没有找到分隔符delim 并且之前在丢弃模式中，由于找不到分隔符，因此仍在丢弃缓冲区，继续进行数据丢弃。
总的来说整个逻辑以三个维度进行判断：有没有找到分隔符、是否处于丢弃模式、有没有超过每帧的最大长度。
如果找到了分隔符，先判断是否处于丢弃模式，如果处于丢弃模式，无论每帧是否大于最大长度，都跳过数据，抛出异常，丢弃模式解除。如果未处于丢弃模式，在判断每帧是否大于最大长度，如果大于最大长度，跳过数据，抛出异常，如果没有大于最长长度，那么久就正常解码数据。
如果未找到分隔符，先判断是否处于丢弃模式，如果处于丢弃模式，继续丢弃数据，如果未在丢弃模式中，判断是否大于最大长度，如果大于最大长度，那么就跳过数据，开启丢弃模式，如果不大于最大长度，那么就返回null，期待下次数据正常解析。
至此，服务端利用DelimiterBasedFrameDecoder解码器解析数据的流程也就分析完了。
数据库
文章转载自文一西路代码狗，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。

评论