dubbo系列之粘包和拆包

前言

其实在前面几篇网络编程的文章里面，关于粘包和拆包已经讲的差不多了，由于本人这周五准备组内分享——dubbo的粘包和拆包，这篇博客就把dubbo处理粘包和拆包的流程梳理总结一下。

什么是粘包和拆包

TCP粘包拆包发生的原因

1.应用程序write写入的字节大小大于套接口发送缓冲区大小

2.进行MSS大小的TCP分段

3.以太网帧的payload(有效数据)大于MTU进行IP分片

以往业界时如何解决粘包和拆包的？

1.消息的定长，例如定1000个字节

2.就是在包尾增加回车或空格等特殊字符作为切割。典型的FTP协议

3.将消息分为消息头消息体，消息头中包含表示消息总长度（或者消息体长度）的字段，通常涉及思路为消息头的第一个字段使用int32来表示消息的总长度。例如 dubbo

dubbo为什么没有选择netty原生的粘包和拆包的方法，选择自定义？

因为dubbo的协议栈是自定义的，里面包含消息头和消息体两个部分，消息头中包含消息类型、协议版本、协议魔数以及playload长度等信息。而netty只能处理一些通用简单的协议栈，并不能处理高度自定义的协议栈，所以dubbo自定义了属于自己的一套粘包和拆包的解决办法。

消息头和消息体长什么样？

以provider给consumer发送的数据包为例

ExchangeCodec类—encodeResponse()

Serialization serialization = getSerialization(channel);
// header.
byte[] header = new byte[HEADER_LENGTH];
// set magic number.
Bytes.short2bytes(MAGIC, header);
// set request and serialization flag.
header[2] = serialization.getContentTypeId();//序列化ID
if (res.isHeartbeat()) header[2] |= FLAG_EVENT;
// set response status.
byte status = res.getStatus();
header[3] = status;//第四个字节为20，代表"ok"
// set request id.
Bytes.long2bytes(res.getId(), header, 4);

int savedWriteIndex = buffer.writerIndex();
buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);
ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer);
ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
// encode response data or error message.
if (status == Response.OK) {
    if (res.isHeartbeat()) {
        encodeHeartbeatData(channel, out, res.getResult());
    } else {
        encodeResponseData(channel, out, res.getResult());
    }
} else out.writeUTF(res.getErrorMessage());
out.flushBuffer();
bos.flush();
bos.close();

int len = bos.writtenBytes();
checkPayload(channel, len);
Bytes.int2bytes(len, header, 12);
// write
buffer.writerIndex(savedWriteIndex);
buffer.writeBytes(header); // write header.
buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);//写入消息头和消息体总长度

dubbo的消息头是一个定长的 16个字节。
第1-2个字节：是一个魔数数字：就是一个固定的数字
第3个字节：序列号组件类型，它用于和客户端约定的序列号编码号
第四个字节：它是response的结果响应码 例如 OK=20
第5-12个字节：请求id：long型8个字节。异步变同步的全局唯一ID，用来做consumer和provider的来回通信标记。
第13-16个字节：消息体的长度，也就是消息头+请求数据的长度。

这样，接着看consumer接受到编码后的消息体，是如何处理的

NettyCodecAdapter.InternalDecoder类

messageReceived()

分两个部分

//首先判断当前decoder对象的buffer中是否有可以读取的消息，若有，则合并。并将对象赋值给message局部变量
if (buffer.readable()) {
    if (buffer instanceof DynamicChannelBuffer) {
        buffer.writeBytes(input.toByteBuffer());
        message = buffer;//message获取当前channel的inbound消息
    } 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 {
           //保存当前message的读索引，用于后期消息回滚
           saveReaderIndex = message.readerIndex();
           try {
           //DubboCountCodec的decode每次只会解析出一个完整的dubbo协议栈
           msg = codec.decode(channel, message);//debug
        } 
   }

DubboCountCodec类

decode()

public Object decode(Channel channel, ChannelBuffer buffer) throws IOException {
	int save = buffer.readerIndex();//保存buffer的读索引
    MultiMessage result = MultiMessage.create();
    do {
        Object obj = codec.decode(channel, buffer);//debug
        if (Codec2.DecodeResult.NEED_MORE_INPUT == obj) {//数据不足
            buffer.readerIndex(save);//将buffer读索引回滚
            break;
        } else {
            result.addMessage(obj);
            logMessageLength(obj, buffer.readerIndex() - save);
            save = buffer.readerIndex();//save更新
        }
    } 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的读索引回滚到之前暂存的索引并且退出循环，将结果返回。

ExchangeCodec类

decode()

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.
    //1：当前buffer的可读长度还没有消息头长，说明当前buffer连协议栈的头都不完整
    if (readable < HEADER_LENGTH) {
        return DecodeResult.NEED_MORE_INPUT;
    }

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

    int tt = len + HEADER_LENGTH;
    //2：当前buffer包含了完整的消息头，便可以得到payload的长度，发现它的可读的长度，并没有包含整个协议栈的数据
    if (readable < tt) {
        return DecodeResult.NEED_MORE_INPUT;
    }

    // limit input stream.
    // 3：如果上面两个情况都不符合，那么说明当前的buffer至少包含一个dubbo协议栈的数据，那么从当前buffer中读取一个dubbo协议栈的数据，解析出一个dubbo数据，当然这里可能读取完一个dubbo数据之后还会有剩余的数据。读取一个dubbo协议栈的数据
    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);
            }
        }
    }
}

##总结

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

上面的方法就解决了粘包了拆包的问题了

若出现粘包，就会根据它的请求长度进行截取；

若出现拆包，数据会不完整，就进入循环重新读取，直到取到完整数据。

END