【Dubbo】编码解码

编解码含义

编码（Encode）
编码它将对象序列化为字节数组，用于网络传输、数据持久化或者其他用途。
解码（Decode）
解码把从网络、磁盘等读取的字节数组还原成原始对象（通常是原始对象的拷贝），以方便后续的业务逻辑操作。

粘包拆包

tcp 为什么会出现粘包拆包的问题？

TCP报文有个比较大的特点，就是它传输的时候，会先把应用层的数据项拆开成字节，然后按照自己的传输需要，选择合适数量的字节进行传输。什么叫"自己的传输需要"？首先TCP包有最大长度限制，那么太大的数据项肯定是要拆开的。其次因为TCP以及下层协议会附加一些协议头信息，如果数据项太小，那么可能报文大部分都是没有价值的头信息，这样传输是很不划算的。
TCP/IP协议与buffer

粘包拆包

解决方案

消息的定长，例如定1000个字节
就是在包尾增加回车或空格等特殊字符作为切割，典型的FTP协议
将消息分为消息头消息体，并记录数据长度。例如 dubbo

netty自带的编解码器

    public ChannelPipeline getPipeline() throws Exception {
        ChannelPipeline channelPipeline = Channels.pipeline();
        channelPipeline.addLast("decoder", new StringDecoder());
        channelPipeline.addLast("encoder", new StringEncoder());
        channelPipeline.addLast("handler", new ServerLogicHandler());
        return channelPipeline;
    }

dubbo自己的编解码协议
出现粘包拆包的核心是需要将TCP层的字节数据包切分业务层的Request、Response数据包。所以dubbo使用了魔法数和数据长度作为分割符，将Buffer中的数据隔离出来了生成Request、Response这样的一个个业务层的对象。

image.png

编解码核心

编解码流程

编解码流程
注册编解码器

#com.alibaba.dubbo.remoting.transport.netty.NettyServer#doOpen
bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
            public ChannelPipeline getPipeline() {
                NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec() ,getUrl(), NettyServer.this);
                ChannelPipeline pipeline = Channels.pipeline();
                /*int idleTimeout = getIdleTimeout();
                if (idleTimeout > 10000) {
                    pipeline.addLast("timer", new IdleStateHandler(timer, idleTimeout / 1000, 0, 0));
                }*/
                //解码
                pipeline.addLast("decoder", adapter.getDecoder());
                //编码
                pipeline.addLast("encoder", adapter.getEncoder());
                //逻辑处理类
                pipeline.addLast("handler", nettyHandler);
                return pipeline;
            }
        });
#com.alibaba.dubbo.remoting.exchange.codec.ExchangeCodec#encode
    public void encode(Channel channel, ChannelBuffer buffer, Object msg) throws IOException {
        if (msg instanceof Request) {
            encodeRequest(channel, buffer, (Request) msg);
        } else if (msg instanceof Response) {
            encodeResponse(channel, buffer, (Response) msg);
        } else {
            super.encode(channel, buffer, msg);
        }
    }

    public Object decode(Channel channel, ChannelBuffer buffer) throws IOException {
        int readable = buffer.readableBytes();
        //读取header，可能可读数据没有完全传输过来
        byte[] header = new byte[Math.min(readable, HEADER_LENGTH)];
        buffer.readBytes(header);
        return decode(channel, buffer, readable, header);
    }

Request编码

----------1------consumer请求编码----------------------
-->NettyCodecAdapter.InternalEncoder.encode
  -->DubboCountCodec.encode
    -->ExchangeCodec.encode
      -->ExchangeCodec.encodeRequest
        -->DubboCodec.encodeRequestData

consumer发起Request是在调用代理对象的方法后会触发DubboInvoker.doInvoke使用netty发送Invocktion数据。编码器将一些Request标志位按照编码规则处理成字节数组，将Invocktion序列化并计算出数据长度存放在data部分，将处理好的字节数组都写入可扩容的Buffer完成编码

#com.alibaba.dubbo.rpc.protocol.dubbo.DubboInvoker#doInvoke
RpcInvocation inv = (RpcInvocation) invocation;
ResponseFuture future = currentClient.request(inv, timeout) ;
                //取得结果的future放在上下文中
RpcContext.getContext().setFuture(new FutureAdapter<Object>(future));

#com.alibaba.dubbo.remoting.exchange.support.header.HeaderExchangeChannel#request(java.lang.Object, int)
 public ResponseFuture request(Object request, int timeout) throws RemotingException {
        if (closed) {
            throw new RemotingException(this.getLocalAddress(), null, "Failed to send request " + request + ", cause: The channel " + this + " is closed!");
        }
        // create request.
        Request req = new Request();
        req.setVersion("2.0.0");
        req.setTwoWay(true);
        req.setData(request);
        DefaultFuture future = new DefaultFuture(channel, req, timeout);
        try{
            //AbstractPeer
            channel.send(req);
        }catch (RemotingException e) {
            future.cancel();
            throw e;
        }
        return future;
    }

消息头
dubbo的消息头是一个定长的 16个字节。
第1-2个字节：是一个魔数数字：就是一个固定的数字
第3个字节：序列化器的id、是否是事件、是双向(有去有回) 或单向（有去无回）的标记
第4个字节：status？？？（request 没有第四个字节）
第5-12个字节：请求id：long型8个字节。异步变同步的全局唯一ID，用来做consumer和provider的来回通信标记。
第13-16个字节：消息体的长度，也就是消息头+请求数据的长度。
消息体
data实际上是一个Invocation，标识了需要调用的方法以及参数。而且使用的buffer是可以动态扩展的

#com.alibaba.dubbo.rpc.protocol.dubbo.DubboInvoker#doInvoke
RpcInvocation inv = (RpcInvocation) invocation;
ResponseFuture future = currentClient.request(inv, timeout) ;

编码流程

#com.alibaba.dubbo.remoting.exchange.codec.ExchangeCodec#encodeRequest
 protected void encodeRequest(Channel channel, ChannelBuffer buffer, Request req) throws IOException {
        Serialization serialization = getSerialization(channel);
        // header.16字节，128位
        byte[] header = new byte[HEADER_LENGTH];
        // set magic number.16位魔法数0xdabb
        //short2bytes 将16位数存到两个byte中
        Bytes.short2bytes(MAGIC, header);

        // set request and serialization flag.
        header[2] = (byte) (FLAG_REQUEST | serialization.getContentTypeId());

        if (req.isTwoWay()) header[2] |= FLAG_TWOWAY;
        if (req.isEvent()) header[2] |= FLAG_EVENT;

        // set request id.
        //long 64位8字节，从第5个字节
        Bytes.long2bytes(req.getId(), header, 4);
        //buffer这里写入buffer并不是直接写入到netty，是一个新建的buffer
        //写完需要netty发送
        // encode request data. 保存当前写入位置
        int savedWriteIndex = buffer.writerIndex();
        buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);
        ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer);
        ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
        if (req.isEvent()) {
            encodeEventData(channel, out, req.getData());
        } else {
            //编码写入请求数据DubboCodec
            encodeRequestData(channel, out, req.getData());
        }
        out.flushBuffer();
        bos.flush();
        bos.close();
        //获取写入数据的长度
        int len = bos.writtenBytes();
        checkPayload(channel, len);
        Bytes.int2bytes(len, header, 12);
        // write
        buffer.writerIndex(savedWriteIndex);
        //写header
        buffer.writeBytes(header); // write header.
        //设置当前写入位置
        buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);
    }

Request解码

provider接收到consumer的Request数据之后会进入Request解码流程

----------2------provider 请求解码----------------------
--NettyCodecAdapter.InternalDecoder.messageReceived
  -->DubboCountCodec.decode
    -->ExchangeCodec.decode
      -->ExchangeCodec.decodeBody

解码流程就是接收到数据之后，先读取header长度的字节。判断开头两字节是不是magicNumber，如果不是则遍历数据的每一位来判断魔法数，交给父类来解码；是则校验header长度，数据长度，根据header的属性构造Response、Request等对象，然后读取序列化器来解析data默认使用Hessian进行序列化和反序列化

#com.alibaba.dubbo.remoting.exchange.codec.ExchangeCodec#decode(com.alibaba.dubbo.remoting.Channel, com.alibaba.dubbo.remoting.buffer.ChannelBuffer)
public Object decode(Channel channel, ChannelBuffer buffer) throws IOException {
        int readable = buffer.readableBytes();
        //读取header
        byte[] header = new byte[Math.min(readable, HEADER_LENGTH)];
        buffer.readBytes(header);
        return decode(channel, buffer, readable, header);
    }
    
    protected Object decode(Channel channel, ChannelBuffer buffer, int readable, byte[] header) throws IOException {
        // check magic number.
        //每次解码先判断是否是magic——number
        if (readable > 0 && header[0] != MAGIC_HIGH 
                || readable > 1 && header[1] != MAGIC_LOW) {
            //数据的起始位置不是header
            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) {
                    //检查到magic
                    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 {
            //body解析
            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);
                }
            }
        }

provider响应结果编码

provider调用完本地的invoker后需要返回Response，返回数据也是需要编码

#com.alibaba.dubbo.remoting.exchange.support.header.HeaderExchangeHandler#received
    public void received(Channel channel, Object message) throws RemotingException {
    if (message instanceof Request) {
                // handle request.
                Request request = (Request) message;
                if (request.isEvent()) {
                    handlerEvent(channel, request);
                } else {
                    if (request.isTwoWay()) {
                        //接收数据网络通信接收处理
                        Response response = handleRequest(exchangeChannel, request);
                        channel.send(response);
                    } else {
                        handler.received(exchangeChannel, request.getData());
                    }
                }
            } else if (message instanceof Response) {
                handleResponse(channel, (Response) message);
            }
}

----------3------provider响应结果编码----------------------
-->NettyCodecAdapter.InternalEncoder.encode
  -->DubboCountCodec.encode
    -->ExchangeCodec.encode
      -->ExchangeCodec.encodeResponse
        -->DubboCodec.encodeResponseData//先写入一个字节 这个字节可能是RESPONSE_NULL_VALUE  RESPONSE_VALUE  RESPONSE_WITH_EXCEPTION

编码协议都是一致的，和consumer请求编码有一小点不同
第3个字节：序列号组件类型，它用于和客户端约定的序列号编码号
第4个字节：它是response的结果响应码例如 OK=20

consumer请求结果解码

----------4------consumer响应结果解码----------------------
--NettyCodecAdapter.InternalDecoder.messageReceived
  -->DubboCountCodec.decode
    -->ExchangeCodec.decode
      -->DubboCodec.decodeBody
        -->DecodeableRpcResult.decode//根据RESPONSE_NULL_VALUE  RESPONSE_VALUE  RESPONSE_WITH_EXCEPTION进行响应的处理

解码完成后是一个Response

解码之后

nettyServer的pipeline中注册三个处理器，前两个用来数据的编解码。在收到的数据已经完成解码工作成为了Request、Response之后就该进入nettyHandler的处理了

pipeline.addLast("decoder", adapter.getDecoder());
                //编码
pipeline.addLast("encoder", adapter.getEncoder());
                //逻辑处理类
pipeline.addLast("handler", nettyHandler);

#com.alibaba.dubbo.remoting.transport.netty.NettyHandler#messageReceived
    @Override
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
        //数据接收
        NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
        try {
            handler.received(channel, e.getMessage());
        } finally {
            NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
        }
    }

DecodeHandler区分数据是Request还是Response，Request中的数据解析为DecodeableRpcInvocation,Response中的数据解析为DecodeableRpcResult

#com.alibaba.dubbo.remoting.transport.DecodeHandler#received
public void received(Channel channel, Object message) throws RemotingException {
        if (message instanceof Decodeable) {
            decode(message);
        }

        if (message instanceof Request) {
            //如果是consumer请求则解析data为invocation
            decode(((Request)message).getData());
        }

        if (message instanceof Response) {
            //解析请求返回的response数据
            decode( ((Response)message).getResult());
        }

        handler.received(channel, message);
    }

如果是Request则寻找本地的invoker并调用并发回返回值Response，如果是Response则处理处理之前保存的future唤醒等待的线程

#com.alibaba.dubbo.remoting.exchange.support.header.HeaderExchangeHandler#received
public void received(Channel channel, Object message) throws RemotingException {
        channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis());
        ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
        try {
            if (message instanceof Request) {
                // handle request.
                Request request = (Request) message;
                if (request.isEvent()) {
                    handlerEvent(channel, request);
                } else {
                    if (request.isTwoWay()) {
                        //有返回值的请求
                        Response response = handleRequest(exchangeChannel, request);
                        channel.send(response);
                    } else {
                        handler.received(exchangeChannel, request.getData());
                    }
                }
            } else if (message instanceof Response) {
                handleResponse(channel, (Response) message);
            } else if (message instanceof String) {
                if (isClientSide(channel)) {
                    Exception e = new Exception("Dubbo client can not supported string message: " + message + " in channel: " + channel + ", url: " + channel.getUrl());
                    logger.error(e.getMessage(), e);
                } else {
                    String echo = handler.telnet(channel, (String) message);
                    if (echo != null && echo.length() > 0) {
                        channel.send(echo);
                    }
                }
            } else {
                handler.received(exchangeChannel, message);
            }
        } finally {
            HeaderExchangeChannel.removeChannelIfDisconnected(channel);
        }
    }
static void handleResponse(Channel channel, Response response) throws RemotingException {
        if (response != null && !response.isHeartbeat()) {
            DefaultFuture.received(channel, response);
        }
    }

解决粘包拆包问题

出现拆包可能没有魔法数或者数据长度不够，那就跳出循环等待下一次数据；如果出现粘包会有多个魔法数每次解码一个循环多次解码。

#com.alibaba.dubbo.rpc.protocol.dubbo.DubboCountCodec
    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;
    }

总结

出现粘包拆包的核心是需要将TCP层的字节数据包切分业务层的Request、Response数据包。所以dubbo使用了魔法数和数据长度作为分割符，将Buffer中的数据隔离出来了生成Request、Response这样的一个个业务层的对象。

【Dubbo】编码解码

编解码含义

粘包 拆包

tcp 为什么会出现粘包 拆包的问题？

解决方案

编解码核心

Request编码

Request解码

provider响应结果编码

consumer请求结果解码

解码之后

解决粘包拆包问题

总结

粘包拆包

tcp 为什么会出现粘包拆包的问题？