前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >Netty 黏包拆包机制

Netty 黏包拆包机制

作者头像
luoxn28
发布2019-12-12 16:48:59
6200
发布2019-12-12 16:48:59
举报
文章被收录于专栏:TopCoderTopCoderTopCoder

编者注:学习netty处理黏包和拆包,首先要知道什么是黏包和拆包问题?

黏包和拆包的产生是由于TCP拥塞控制算法(比如angle算法)和TCP缓冲区机制导致的,angle算法简单来说就是通过一些规则来尽可能利用网络带宽,尽可能的发送足够大的数据。TCP(发送/接收)缓冲区会暂缓数据,并且是有最大容量的。

黏包的产生是由于一次TCP通信数据量较少,导致多个TCP数据合并在一起(这里的合并可能发生在发送缓冲区合并后发送,也可能发生在接收缓冲区合并后应用程序一次性读取)。拆包的产生是由于一次TCP通信数据量较大(比如超过了MTU),导致发送时分片发送,这样接收时是多次接收后才是一个完整的数据。

netty处理黏包和拆包问题,思路就是以定长方式读取接收到的数据来处理(比如读取固定长度数据量、以TLV方式读取数据、以固定分隔符读取数据等)

下面以一个定长方式读取数据的示例来分析下Netty的处理机制,server端处理TCP黏包和拆包示例代码:

EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
    ServerBootstrap boot = new ServerBootstrap();
    boot.group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class)
            .localAddress(60000)
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline()
                            // 定长接收消息,用于处理黏包分包问题
                            .addLast(new FixedLengthFrameDecoder(1))
                            .addLast(new EchoHandler());
                }
            });

    // start
    ChannelFuture future = boot.bind().sync();
    future.channel().closeFuture().sync();
} catch (Exception e) {
    e.printStackTrace();
} finally {
    // shutdown
    bossGroup.shutdownGracefully();
    workerGroup.shutdownGracefully();
}

FixedLengthFrameDecoder类继承关系如下,下面就以该类为例讲解下netty 处理粘包分包机制。

这里思考一下,如果接收到的数据未达到FixedLengthFrameDecoder要求的长度,这个时候Netty该如何处理呢?

首先可以肯定的一点是,接收到数据长度不够,是不会进行后续channelHandler处理的。Netty的处理机制是会将接收到的数据存储到ByteToMessageDecoder.cumulation中,暂存一下,等待下次接收到数据时继续处理直到达到要求长度之后才交给后续的ChannelHandler来处理,ByteToMessageDecoder代码如下:

// ByteToMessageDecoder
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保存有上次未处理(不完整)的数据
                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) {
                numReads = 0;
                discardSomeReadBytes();
            }

            int size = out.size();
            decodeWasNull = !out.insertSinceRecycled();
            // 继续回调channelHandler.channelRead
            fireChannelRead(ctx, out, size);
            out.recycle();
        }
    } else {
        ctx.fireChannelRead(msg);
    }
}

final void decodeRemovalReentryProtection(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
        throws Exception {
    decodeState = STATE_CALLING_CHILD_DECODE;
    try {
        decode(ctx, in, out);
    } finally {
        boolean removePending = decodeState == STATE_HANDLER_REMOVED_PENDING;
        decodeState = STATE_INIT;
        if (removePending) {
            handlerRemoved(ctx);
        }
    }
}
protected final void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
    Object decoded = decode(ctx, in);
    if (decoded != null) {
        out.add(decoded);
    }
}
// 已接收数据未达到要求的长度时,继续等待接收
protected Object decode(
        @SuppressWarnings("UnusedParameters") ChannelHandlerContext ctx, ByteBuf in) throws Exception {
    if (in.readableBytes() < frameLength) {
        return null;
    } else {
        return in.readRetainedSlice(frameLength);
    }
}

处理粘包拆包,其实思路都是一致的,就是“分分合合”,粘包由于数据过多,那就按照固定策略分割下交给程序来处理;拆包由于一次传输数据较少,那就等待数据传输长度够了再交给程序来处理。

本文参与 腾讯云自媒体分享计划,分享自微信公众号。
原始发表:2019-11-30,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 TopCoder 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体分享计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
相关产品与服务
数据保险箱
数据保险箱(Cloud Data Coffer Service,CDCS)为您提供更高安全系数的企业核心数据存储服务。您可以通过自定义过期天数的方法删除数据,避免误删带来的损害,还可以将数据跨地域存储,防止一些不可抗因素导致的数据丢失。数据保险箱支持通过控制台、API 等多样化方式快速简单接入,实现海量数据的存储管理。您可以使用数据保险箱对文件数据进行上传、下载,最终实现数据的安全存储和提取。
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档