`

Dubbo处理TCP拆包粘包问题

    博客分类：
  • Java
 
阅读更多
摘要 在TCP网络传输工程中，由于TCP包的缓存大小限制，每次请求数据有可能不在一个TCP包里面，或者也可能多个请求的数据在一个TCP包里面。那么如果合理的decode接受的TCP数据很重要，需要考虑TCP拆包和粘包的问题。我们知道在Netty提供了各种Decoder来解决此类问题，比如`LineBasedFrameDecoder`,`LengthFieldBasedFrameDecoder`等等，但是这些都是处理一些通用简单的协议栈，并不能处理高度自定义的协议栈。由于dubbo协议是自定义协议栈，并且包含消息头和消息体两部分，而消息头中包含消息类型、协议版本、协议魔数以及playload长度...

 

目录[-]

Dubbo处理TCP拆包粘包问题

在TCP网络传输工程中，由于TCP包的缓存大小限制，每次请求数据有可能不在一个TCP包里面，或者也可能多个请求的数据在一个TCP包里面。那么如果合理的decode接受的TCP数据很重要，需要考虑TCP拆包和粘包的问题。我们知道在Netty提供了各种Decoder来解决此类问题，比如LineBasedFrameDecoder,LengthFieldBasedFrameDecoder等等，但是这些都是处理一些通用简单的协议栈，并不能处理高度自定义的协议栈。由于dubbo协议是自定义协议栈，并且包含消息头和消息体两部分，而消息头中包含消息类型、协议版本、协议魔数以及playload长度等信息。所以使用Netty自带的处理方案可能无法满足Dubbo解析自身协议的需求，所以需要Dubbo自己来处理，那自己处理，就需要自己处理TCP的拆包和粘包的问题。这里就对Dubbo处理此类问题进行探讨，从而加深自己对它的理解。

说明

此处所描述的协议是dubbo协议，其他的协议比如http，webservice等协议不是这里讨论范围。并且这里使用的通信框架以Netty来讲解，Mina以及grizzly也不在种类讨论范围。

NettyCodecAdapter

NettyCodecAdapter是对dubbo协议解析的入口，里面包含decoder和encoder两部分，而TCP的拆包和粘包主要是decoder部分，所以encoder这里不进行讨论。在NettyCodecAdapter中的decoder是由InternalDecoder来实现，它的父类是Netty的SimpleChannelUpstreamHandler可以接受所有inbound消息，那么就可以对接受的消息进行decode。这里需要说明一下对于某一个Channel都有一个私有的InternalDecoder对象，并不是和其他的Channel共享，这里就避免了并发问题，所以在InternalDecoder里面可以用单线程的方式去看待，这样就比较容易理解。

InternalDecoder

每个channel的inbound消息都会发送到InternalDecoder的messageReceived方法，而dubbo会先将接受的消息缓存到InternalDecoder的buffer属性中，这个变量很重要，后面会讨论。下面是messageReceived方法中将接受的消息负载到buffer实现。

<!--lang:java-->
  private classInternalDecoderextendsSimpleChannelUpstreamHandler {

    private com.alibaba.dubbo.remoting.buffer.ChannelBuffer buffer =
        com.alibaba.dubbo.remoting.buffer.ChannelBuffers.EMPTY_BUFFER;

    @Override
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent event) throws Exception {
        Object o = event.getMessage();
        if (! (o instanceof ChannelBuffer)) {
            ctx.sendUpstream(event);
            return;
        }

        ChannelBuffer input = (ChannelBuffer) o;
        int readable = input.readableBytes();
        if (readable <= 0) {
            return;
        }

        com.alibaba.dubbo.remoting.buffer.ChannelBuffer message;
        if (buffer.readable()) {
            if (buffer instanceof DynamicChannelBuffer) {
                buffer.writeBytes(input.toByteBuffer());
                message = buffer;
            } else {
                int size = buffer.readableBytes() + input.readableBytes();
                message = com.alibaba.dubbo.remoting.buffer.ChannelBuffers.dynamicBuffer(
                    size > bufferSize ? size : bufferSize);
                message.writeBytes(buffer, buffer.readableBytes());
                message.writeBytes(input.toByteBuffer());
            }
        } else {
            message = com.alibaba.dubbo.remoting.buffer.ChannelBuffers.wrappedBuffer(
                input.toByteBuffer());
        }

        NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
        Object msg;
        int saveReaderIndex;

        try {
            // decode object.
            do {
                saveReaderIndex = message.readerIndex();
                try {
                    msg = codec.decode(channel, message);
                } catch (IOException e) {
                    buffer = com.alibaba.dubbo.remoting.buffer.ChannelBuffers.EMPTY_BUFFER;
                    throw e;
                }
                if (msg == Codec2.DecodeResult.NEED_MORE_INPUT) {
                    message.readerIndex(saveReaderIndex);
                    break;
                } else {
                    if (saveReaderIndex == message.readerIndex()) {
                        buffer = com.alibaba.dubbo.remoting.buffer.ChannelBuffers.EMPTY_BUFFER;
                        throw new IOException("Decode without read data.");
                    }
                    if (msg != null) {
                        Channels.fireMessageReceived(ctx, msg, event.getRemoteAddress());
                    }
                }
            } while (message.readable());
        } finally {
            if (message.readable()) {
                message.discardReadBytes();
                buffer = message;
            } else {
                buffer = com.alibaba.dubbo.remoting.buffer.ChannelBuffers.EMPTY_BUFFER;
            }
            NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
        ctx.sendUpstream(e);
    }
}

首先是判断当前decoder对象的buffer中是否有可以读取的消息，如果有则进行合并，并且把对象引用赋予message局部变量，所以message则获取了当前channel的inbound消息。得到inbound消息之后，那么接下来就是对协议的解析了。

<!--lang:java-->
 do {
                saveReaderIndex = message.readerIndex();
                try {
                    msg = codec.decode(channel, message);
                } catch (IOException e) {
                    buffer = com.alibaba.dubbo.remoting.buffer.ChannelBuffers.EMPTY_BUFFER;
                    throw e;
                }
                if (msg == Codec2.DecodeResult.NEED_MORE_INPUT) {
                    message.readerIndex(saveReaderIndex);
                    break;
                } else {
                    if (saveReaderIndex == message.readerIndex()) {
                        buffer = com.alibaba.dubbo.remoting.buffer.ChannelBuffers.EMPTY_BUFFER;
                        throw new IOException("Decode without read data.");
                    }
                    if (msg != null) {
                        Channels.fireMessageReceived(ctx, msg, event.getRemoteAddress());
                    }
                }
            } while (message.readable());

这里首先要做的是把当前message的读索引保存到局部变量saveReaderIndex中，用于后面的消息回滚。后面紧接着是对消息的decode，这里的codec是DubboCountCodec对象实体，这里需要注意一点，DubboCountCodec的decode每次只会解析出一个完整的dubbo协议栈，带着这个看看decode的实现。

<!--lang:java-->
public Object decode(Channel channel, ChannelBuffer buffer) throws IOException {
    int save = buffer.readerIndex();
    MultiMessage result = MultiMessage.create();
    do {
        Object obj = codec.decode(channel, buffer);
        if (Codec2.DecodeResult.NEED_MORE_INPUT == obj) {
            buffer.readerIndex(save);
            break;
        } else {
            result.addMessage(obj);
            logMessageLength(obj, buffer.readerIndex() - save);
            save = buffer.readerIndex();
        }
    } while (true);
    if (result.isEmpty()) {
        return Codec2.DecodeResult.NEED_MORE_INPUT;
    }
    if (result.size() == 1) {
        return result.get(0);
    }
    return result;
}

这里暂存了当前buffer的读索引，同样也是为了后面的回滚。可以看到当decode返回的是NEED_MORE_INPUT则表示当前的buffer中数据不足，不能完整解析出一个dubbo协议栈，同时将buffer的读索引回滚到之前暂存的索引并且退出循环，将结果返回。那接下来看看什么时候会返回NEED_MORE_INPUT，最终会定位到在ExchangeCodec的decode方法会解析出协议栈。

<!--lang:java-->
protected Object decode(Channel channel, ChannelBuffer buffer, int readable, byte[] header) throws IOException {
    // check magic number.
    if (readable > 0 && header[0] != MAGIC_HIGH 
            || readable > 1 && header[1] != MAGIC_LOW) {
        int length = header.length;
        if (header.length < readable) {
            header = Bytes.copyOf(header, readable);
            buffer.readBytes(header, length, readable - length);
        }
        for (int i = 1; i < header.length - 1; i ++) {
            if (header[i] == MAGIC_HIGH && header[i + 1] == MAGIC_LOW) {
                buffer.readerIndex(buffer.readerIndex() - header.length + i);
                header = Bytes.copyOf(header, i);
                break;
            }
        }
        return super.decode(channel, buffer, readable, header);
    }
    // check length.
    if (readable < HEADER_LENGTH) {
        return DecodeResult.NEED_MORE_INPUT;
    }

    // get data length.
    int len = Bytes.bytes2int(header, 12);
    checkPayload(channel, len);

    int tt = len + HEADER_LENGTH;
    if( readable < tt ) {
        return DecodeResult.NEED_MORE_INPUT;
    }

    // limit input stream.
    ChannelBufferInputStream is = new ChannelBufferInputStream(buffer, len);

    try {
        return decodeBody(channel, is, header);
    } finally {
        if (is.available() > 0) {
            try {
                if (logger.isWarnEnabled()) {
                    logger.warn("Skip input stream " + is.available());
                }
                StreamUtils.skipUnusedStream(is);
            } catch (IOException e) {
                logger.warn(e.getMessage(), e);
            }
        }
    }
}

这个方法开始是对telnet协议进行解析（由于dubbo支持telnet连接，所以这里提供了支持，可以忽略这一部分）。看到会有两个地方返回NEED_MORE_INPUT，一个是当前buffer的可读长度还没有消息头长，说明当前buffer连协议栈的头都不完整，所以需要继续读取inbound数据，另一个是当前buffer包含了完整的消息头，便可以得到playload的长度，发现它的可读的长度，并没有包含整个协议栈的数据，所以也需要继续读取inbound数据。如果上面两个情况都不复核，那么说明当前的buffer至少包含一个dubbo协议栈的数据，那么从当前buffer中读取一个dubbo协议栈的数据，解析出一个dubbo数据，当然这里可能读取完一个dubbo数据之后还会有剩余的数据。

上面对dubbo解析出一个完整的dubbo协议栈过程进行了讨论，但是还没有对TCP的拆包和粘包问题做过多的讨论。下面结合上面内容做一个综合讨论。

我这里对TCP拆包和粘包分别列举一个场景来讨论。

当反生TCP拆包问题时候

这里假设之前还没有发生过任何数据交互，系统刚刚初始化好，那么这个时候在InternalDecoder里面的buffer属性会是EMPTY_BUFFER。当发生第一次inbound数据的时候，第一次在InternalDecoder里面接收的肯定是dubbo消息头的部分（这个由TCP协议保证），由于发生了拆包情况，那么此时接收的inbound消息可能存在一下几种情况

1、当前inbound消息只包含dubbo协议头的一部分

2、当前inbound消息只包含dubbo的协议头

3、当前inbound消息只包含dubbo消息头和部分playload消息

通过上面的讨论，我们知道发生上面三种情况，都会触发ExchangeCodec返回NEED_MORE_INPUT，由于在DubboCountCodec对余返回NEED_MORE_INPUT会回滚读索引，所以此时的buffer里面的数据可以当作并没有发生过读取操作，并且DubboCountCodec的decode也会返回NEED_MORE_INPUT，在InternalDecoder对于当判断返回NEED_MORE_INPUT，也会进行读索引回滚，并且退出循环，最后会执行finally内容，这里会判断inbound消息是否还有可读的，由于在DubboCountCodec里面进行了读索引回滚，所以次数的buffer里面是完整的inbound消息，等待第二次的inbound消息的到来，当第二次inbound消息过来的时候，再次经过上面的判断。

当发生TCP粘包的时候

当发生粘包的时候是tcp将一个以上的dubbo协议栈放在一个tcp包中，那么有可能发生下面几种情况

1、当前inbound消息只包含一个dubbo协议栈

2、当前inbound消息包含一个dubbo协议栈，同时包含部分另一个或者多个dubbo协议栈内容

如果发生只包含一个协议栈，那么当前buffer通过ExchangeCodec解析协议之后，当前的buffer的readeIndex位置应该是buffer尾部，那么在返回到InternalDecoder中message的方法readable返回的是false,那么就会对buffer重新赋予EMPTY_BUFFER实体，而针对包含一个以上的dubbo协议栈，当然也会解析出其中一个dubbo协议栈，但是经过ExchangeCodec解析之后，message的readIndex不在message尾部，所以message的readable方法返回的是true。那么则会继续遍历message，读取下面的信息。最终要么message刚好整数倍包含完整的dubbo协议栈，要不ExchangeCodec返回NEED_MORE_INPUT,最后将未读完的数据缓存到buffer中,等待下次inbound事件，将buffer中的消息合并到下次的inbound消息中，种类又回到了拆包的问题上。

总结

dubbo在处理tcp的粘包和拆包时是借助InternalDecoder的buffer缓存对象来缓存不完整的dubbo协议栈数据，等待下次inbound事件，合并进去。所以说在dubbo中解决TCP拆包和粘包的时候是通过buffer变量来解决的。

分享到：
| windows下如何github ssh 公钥
评论
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

    我是一个Dubbo数据包....doc

    - NettyCodecAdapter的InternalDecoder处理拆包和粘包问题，确保数据的正确性。 - 解码后的数据包将经过Filter链，最终达到业务方法进行处理。 4. **关键技术点** - **线程模型**：Dubbo中的线程模型包括业务...

    基于Dubbo的跨语言服务治理框架概述.pdf

    4. **粘包和半包处理**：Dubbo通过请求ID和消息体长度来处理TCP层的粘包和拆包问题，借助InternalDecoder的缓冲区处理不完整的数据。 5. **负载均衡策略**：Dubbo提供了多种负载均衡策略，如RandomLoadBalance...

    Netty源码解析1

    5. 研究编解码器，了解如何处理拆包和粘包问题。 6. 学习Netty的线程模型，包括Reactor模式的实现。 7. 通过实践编写简单的Netty应用，逐步加深理解。 对于初学者，可以按照总分总的结构，将Netty的知识点拆分为...

    Netty 17道面试题和答案.docx

    3. 内置的编解码器，处理TCP粘包和拆包问题，简化了协议解析。 4. 预定义的协议栈，支持快速实现各种网络协议。 5. 支持无连接的数据包套接字，适合UDP等无状态通信。 6. 更高的吞吐量、更低的延迟和资源消耗，以及...

    Netty 25道面试题你和答案.docx

    在处理TCP粘包和拆包问题时，Netty 提供了解码器如FixedLengthFrameDecoder，使得开发者可以轻松解决此类问题。 Netty 在各种场景下都有广泛的应用，包括构建高性能的Java中间件（如RocketMQ和Dubbo）、基础通信...

    【BAT必备】netty面试题

    - **使用协议编码器和解码器：**Netty提供了多种编码器和解码器来处理粘包和拆包问题。 通过以上对Netty的基础概念、核心原理、应用场景以及进阶话题的详细介绍，希望能帮助准备面试的朋友们更好地理解和掌握Netty...

    EasyRpc:EasyRpc是一个基于Netty，ZooKeeper和ProtoStuff的简单，高性能，易于使用的RPC框架。

    解决TCP粘包/拆包问题；支持非双向的同步/异步调用；基于ProtoStuff的对象序列化；完整的单元测试和JMH性能压测；基于ZooKeeper实现的服务注册和发现；仿Dubbo数据包结构，优化协议头仅20字节；支持4种负载均衡策略...

    JAVA面试题2019

    4. **粘包/拆包问题**：TCP协议中可能出现的粘包和拆包问题，以及Netty提供的解决方案。 5. **序列化协议**：Netty支持的序列化协议及其适用场景，如JSON、Protobuf等。 6. **零拷贝技术**：Netty如何实现高效的数据...

    Netty面试题（2022最新版）

    TCP粘包/拆包的原因及解决方法 - **原因**：TCP是流式传输，没有消息长度的概念。 - **解决方法**： - 使用固定长度的消息格式； - 在消息之间添加分隔符； - 使用自定义协议，包含消息长度等元信息； - 利用...

    java程序员面试大纲错过了金三银四你还要错过2018吗.docx

    4. **TCP粘包/拆包**：由于TCP协议的流式传输特性，可能导致数据包边界模糊不清，通过设置合理的发送间隔或使用特殊分隔符等方法可以有效避免该问题。 5. **序列化协议**：如JSON、Protobuf、Thrift等，选择合适的...

Magicbox
Global site tag (gtag.js) - Google Analytics