文档章节

Dubbo处理TCP拆包粘包问题

9
 9830ying
发布于 2017/04/14 12:53
字数 2549
阅读 55
收藏 0

Dubbo处理TCP拆包粘包问题

在TCP网络传输工程中,由于TCP包的缓存大小限制,每次请求数据有可能不在一个TCP包里面,或者也可能多个请求的数据在一个TCP包里面。那么如果合理的decode接受的TCP数据很重要,需要考虑TCP拆包和粘包的问题。我们知道在Netty提供了各种Decoder来解决此类问题,比如LineBasedFrameDecoder,LengthFieldBasedFrameDecoder等等,但是这些都是处理一些通用简单的协议栈,并不能处理高度自定义的协议栈。由于dubbo协议是自定义协议栈,并且包含消息头和消息体两部分,而消息头中包含消息类型、协议版本、协议魔数以及payload长度等信息。所以使用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消息都会发送到InternalDecodermessageReceived方法,而dubbo会先将接受的消息缓存到InternalDecoderbuffer属性中,这个变量很重要,后面会讨论。下面是messageReceived方法中将接受的消息负载到buffer实现。

      private class InternalDecoder extends SimpleChannelUpstreamHandler {

        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消息之后,那么接下来就是对协议的解析了。

     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,这里的codecDubboCountCodec对象实体,这里需要注意一点,DubboCountCodecdecode每次只会解析出一个完整的dubbo协议栈,带着这个看看decode的实现。

    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,最终会定位到在ExchangeCodecdecode方法会解析出协议栈。

    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包含了完整的消息头,便可以得到payload的长度,发现它的可读的长度,并没有包含整个协议栈的数据,所以也需要继续读取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消息头和部分payload消息

通过上面的讨论,我们知道发生上面三种情况,都会触发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尾部,那么在返回到InternalDecodermessage的方法readable返回的是false,那么就会对buffer重新赋予EMPTY_BUFFER实体,而针对包含一个以上的dubbo协议栈,当然也会解析出其中一个dubbo协议栈,但是经过ExchangeCodec解析之后,message的readIndex不在message尾部,所以messagereadable方法返回的是true。那么则会继续遍历message,读取下面的信息。最终要么message刚好整数倍包含完整的dubbo协议栈,要不ExchangeCodec返回NEED_MORE_INPUT,最后将未读完的数据缓存到buffer中,等待下次inbound事件,将buffer中的消息合并到下次的inbound消息中,种类又回到了拆包的问题上。

总结

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

源码来源参考:minglisoft.cn/technology欢迎参考交流

© 著作权归作者所有

9
粉丝 1
博文 18
码字总数 49797
作品 0
深圳
私信 提问

暂无文章

聊聊nacos的notifyConfigInfo

序 本文主要研究一下nacos的notifyConfigInfo CommunicationController nacos-1.1.3/config/src/main/java/com/alibaba/nacos/config/server/controller/CommunicationController.java @Cont......

go4it
昨天
2
0
队列-顺序(c/c++实现)

队列是在线性表功能稍作一改形成的,在生活中排队是不能插队的吧,先排队先得到对待,慢来得排在最后面,这样来就形成了”先进先出“的队列。 队列(顺序)实现的三个步骤: 定义所需的功能(...

白客C
昨天
5
0
安装tomcat部署网站/nginx代理tomcat/tomcat部署多个网站

来源:https://blog.csdn.net/Powerful_Fy 安装jdk: Oracle官方下载 https://www.oracle.com/technetwork/java/javase/downloads/index.html yum安装openjdk: 1.查找可使用的jdk包: [roo......

asnfuy
昨天
4
0
Linux 安装 Android studio

官网下载linux 版本 android xxxxx-xxxx.tar.gz 然后 找到它 解压缩 得到 android-studio文件夹 执行 [root@monkey java]# mv ./android-studio /opt [root@monkey java]# cd /opt/android-s......

MrLins
昨天
6
0
VMware vSphere虚拟机的迁移

VMware vSphere 6.7下,虚拟机的迁移按照移动的内容,可以分为4种: 1)冷迁移(cold migration) 2)热迁移(vMotion) 3)存储热迁移(storage vMotion) 4)完全热迁移(shared-nothing ...

大别阿郎
昨天
4
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部