dubbo通信消息解析过程分析(1)

由于rpc底层涉及网络编程接口,线程模型,网络数据结构,服务协议,细到字节的处理。牵涉内容较多,今天就先从一个点说起。 说说,dubbo通过netty框架做传输层,从接到数据字节流到把字节转换为dubbo上层可读的Request消息对象的过程。当前dubbo还支持mina,grizzly做底层传输层。 这里包括两部,反序列化和解码。我打算分两篇写。这篇主要是说解码的过程。 就是下面这个dubbo架构图中,红框中的部分。

既然是netty做传输层,netty的基础得提一点。 netty框架是通过管道(ChannelPipeline)模型处理网络数据流的,每个管道中有多个处理接点(ChannelHandler), 节点分为,进站(client请求进服务端口)和出站(请求响应出服务端口)两种。比如一个进站消息总是,顺序的(顺序是程序中编码指定的)通过进站处理节点。 同理出站消息,总是顺序的通过出站节点到达网络接口。

dubbo2.5.6版本,传输层dubbo提供有netty3和netty4两种实现,初始化netty通道都在NettyServer类里,两个类同名,包名不同。

具体,netty3在NettyServer类里doOpen()方法:

protected void doOpen() throws Throwable {
        NettyHelper.setNettyLoggerFactory();
        ExecutorService boss = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerBoss", true));
        ExecutorService worker = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerWorker", true));
        ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS));
        bootstrap = new ServerBootstrap(channelFactory);

        final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
        channels = nettyHandler.getChannels();
        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());//设置解码hander
                pipeline.addLast("encoder", adapter.getEncoder());//设置编码hander
                pipeline.addLast("handler", nettyHandler);//自定义NettyHandler 扩展自netty双向handler基类,可以接受进站和出站数据流
                return pipeline;
		//进站的请求,先经过adapter.getDecoder()handler处理,再由nettyHandler处理
		//出站的请求,先经过nettyHandler处理 再由adapter.getEncoder()handler处理
            }
        });
        // bind
        channel = bootstrap.bind(getBindAddress());
    }

netty4版本NettyServer类里doOpen()方法:

  protected void doOpen() throws Throwable {
        NettyHelper.setNettyLoggerFactory();

        bootstrap = new ServerBootstrap();
        //acceptor 事件循环线程
        bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("NettyServerBoss", true));
        //client channel事件循环线程
        workerGroup = new NioEventLoopGroup(getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS),
                new DefaultThreadFactory("NettyServerWorker", true));

        final NettyServerHandler nettyServerHandler = new NettyServerHandler(getUrl(), this);
        channels = nettyServerHandler.getChannels();

        bootstrap.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE)
                .childOption(ChannelOption.SO_REUSEADDR, Boolean.TRUE)
                .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        //编解码器的初始化 
                        NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this);
                        ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug
                                .addLast("decoder", adapter.getDecoder())//设置解码hander
                                .addLast("encoder", adapter.getEncoder())//设置编码hander
                                .addLast("handler", nettyServerHandler);//自定义NettyServerHandler 扩展netty双向handler基类,可以接受进站和出站数据流
                        //进站的请求,先经过adapter.getDecoder()handler处理,再由nettyServerHandler处理
                        //出站的请求,先经过nettyServerHandler处理 再由adapter.getEncoder()handler处理
                    }
                });
        // bind
        ChannelFuture channelFuture = bootstrap.bind(getBindAddress());
        channelFuture.syncUninterruptibly();
        channel = channelFuture.channel();

    }

本次说的是进站流程,请求数据解析成Request对象过程,通过上面的代码和netty特性可知, 进站数据先通过解码hander,经解码成消息Request对象后,再到自定义handler,然后由自定义hanlder通过装饰模式,调用实际服务。

先看下解码handler的实现: 由adapter.getDecoder()这句跟踪到NettyCodecAdapter类的getDecoder()方法 public ChannelHandler getDecoder() { return decoder; } 可以看到,这个方法获取的解码handler,decoder,是NettyCodecAdapter类的私有属性

private final ChannelHandler decoder = new InternalDecoder();

看下InternalDecoder类定义,netty3版本:

     * 这里需要些netty的知识,继承SimpleChannelUpstreamHandler,表明它是进站handler
     * 所以进站的数据流,都会经过本handler对象,具体就是messageReceived方法。
     */
    private class InternalDecoder extends SimpleChannelUpstreamHandler {

        private com.alibaba.dubbo.remoting.buffer.ChannelBuffer buffer =
                com.alibaba.dubbo.remoting.buffer.ChannelBuffers.EMPTY_BUFFER;//这是dubbo根据nio自己实现的buffer

        @Override
        public void messageReceived(ChannelHandlerContext ctx, MessageEvent event) throws Exception {
            Object o = event.getMessage();
            if (!(o instanceof ChannelBuffer)) {//这里ChnnelBuffer是netty基于jdk nio ByteBuffer 实现
                ctx.sendUpstream(event);
                return;
            }

            ChannelBuffer input = (ChannelBuffer) o;
            int readable = input.readableBytes();//到这就是从netty event对象取数据的过程。
            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);//bufferSize 不指定是8k,这里表示最小8K
                    message.writeBytes(buffer, buffer.readableBytes());
                    message.writeBytes(input.toByteBuffer());//把netty 读到的字节流写入message
                }
            } else {//直接通过构造器,构造message
                message = com.alibaba.dubbo.remoting.buffer.ChannelBuffers.wrappedBuffer(
                        input.toByteBuffer());
            }

            NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
            Object msg;
            int saveReaderIndex;
            //从netty框架,读取到的数据,放入message后,下面就是针对message的反序列化和解码过程。
            try {
                // decode object.
                do {
                    saveReaderIndex = message.readerIndex();
                    try {
                        //解码,这里面包括的反序列化
                        msg = codec.decode(channel, message);//重要!!!通过具体编解码实例codec完成解码
                    } catch (IOException e) {
                        buffer = com.alibaba.dubbo.remoting.buffer.ChannelBuffers.EMPTY_BUFFER;
                        throw e;
                    }
                    if (msg == Codec2.DecodeResult.NEED_MORE_INPUT) {//如果解码结果是,Codec2.DecodeResult.NEED_MORE_INPUT,表示,需要更多数据
                        message.readerIndex(saveReaderIndex);//很重要,设置readerIndex为,解码读取前的位置,为了下次再从头读取。
                        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) {//解码完成,这里的msg已经是Request对象。!!
                            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);//向下一个传递。一般在最后一个handler处理异常
        }
    }

netty4版本:

/***
     * netty4 扩展了ByteToMessageDecoder
     * 重写decode 方法,解码完成后,不用像netty3手动Channels.fireMessageReceived 发送事件,
     * netty4自动把对象,传递到下一个handler
     */
    private class InternalDecoder extends ByteToMessageDecoder {

        protected void decode(ChannelHandlerContext ctx, ByteBuf input, List<Object> out) throws Exception {

            ChannelBuffer message = new NettyBackedChannelBuffer(input);

            NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler);

            Object msg;

            int saveReaderIndex;

            try {
                // decode object.
                do {
                    //保存初始,读取位置
                    saveReaderIndex = message.readerIndex();
                    try {
                        msg = codec.decode(channel, message);//重要!!!通过具体编解码实例codec完成解码
                    } catch (IOException e) {
                        throw e;
                    }
                    if (msg == Codec2.DecodeResult.NEED_MORE_INPUT) {
                        //解码失败后,重新设置readerIndex为读取前位置
                        message.readerIndex(saveReaderIndex);
                        break;
                    } else {
                        //is it possible to go here ?
                        if (saveReaderIndex == message.readerIndex()) {
                            throw new IOException("Decode without read data.");
                        }
                        if (msg != null) {
                            //解码成功后,加入到out list中,传递到下一个处理handler
                            out.add(msg);
                        }
                    }
                } while (message.readable());
            } finally {
                NettyChannel.removeChannelIfDisconnected(ctx.channel());
            }
        }
    }

可以看到,无论netty3,还是netty4都是通过,NettyCodecAdapter的codec属性完成解码的, 这里有个概念,编解码handler是通过编解码实例完成编解码的,这里的编解码实例就是codec 而codec实例是由它构造函数从上层方法传递的。如下

public NettyCodecAdapter(Codec2 codec, URL url, com.alibaba.dubbo.remoting.ChannelHandler handler) {
        this.codec = codec;
        this.url = url;
        this.handler = handler;
        int b = url.getPositiveParameter(Constants.BUFFER_KEY, Constants.DEFAULT_BUFFER_SIZE);
        this.bufferSize = b >= Constants.MIN_BUFFER_SIZE && b <= Constants.MAX_BUFFER_SIZE ? b : Constants.DEFAULT_BUFFER_SIZE;
    }

再回到NettySever类中NettyCodecAdapter的构造语句 NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this); 跟进getCodec()方法,这个方就是获取实际编解码方案的。这个方法的实现在NettyServer的祖先类AbstractEndpoint中:

    protected Codec2 getCodec() {
        return codec;
    }
    //可以看到codec在构造方法里创建的
     public AbstractEndpoint(URL url, ChannelHandler handler) {
        super(url, handler);
        this.codec = getChannelCodec(url);//根据url配置,构造编码解码器(通过spi得到DubboCountCodec类实例)
        this.timeout = url.getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
        this.connectTimeout = url.getPositiveParameter(Constants.CONNECT_TIMEOUT_KEY, Constants.DEFAULT_CONNECT_TIMEOUT);
    }
   // 再跟到getChannelCodec方法
     protected static Codec2 getChannelCodec(URL url) {
        //通过spi机制,从url里获取编解码方案,这里是dubbo。取不到就是telnet
	//dubbo编解码方案,实现类是DubboCountCodec
        String codecName = url.getParameter(Constants.CODEC_KEY, "telnet");
        if (ExtensionLoader.getExtensionLoader(Codec2.class).hasExtension(codecName)) {
            return ExtensionLoader.getExtensionLoader(Codec2.class).getExtension(codecName);
        } else {
            return new CodecAdapter(ExtensionLoader.getExtensionLoader(Codec.class)
                    .getExtension(codecName));
        }
    }

这里看下,dubbo中的所有编解码类结构:

可以看到,所有编解码器实现,都扩展了Codec2接口。同时Codec2也是个spi扩展点。 接口Codec2,如下:

@SPI
public interface Codec2 {
    /**
     * spi 获取编码器
     * @param channel
     * @param buffer
     * @param message
     * @throws IOException
     */
    @Adaptive({Constants.CODEC_KEY})
    void encode(Channel channel, ChannelBuffer buffer, Object message) throws IOException;

    /***
     * spi 获取解码器
     * @param channel
     * @param buffer
     * @return
     * @throws IOException
     */
    @Adaptive({Constants.CODEC_KEY})
    Object decode(Channel channel, ChannelBuffer buffer) throws IOException;


    enum DecodeResult {
        NEED_MORE_INPUT, SKIP_SOME_INPUT
    }

}

具体实现通过spi获取,dubbo编解码方案实例就是DubboCountCodec 那么看下DubboCountCodec类,以及decode方法:

public final class DubboCountCodec implements Codec2 {

    private DubboCodec codec = new DubboCodec();//具体dubbo协议编解码方案实现

    public void encode(Channel channel, ChannelBuffer buffer, Object msg) throws IOException {
        codec.encode(channel, buffer, msg);
    }

    public Object decode(Channel channel, ChannelBuffer buffer) throws IOException {
        int save = buffer.readerIndex();//记录读取初始位置
        MultiMessage result = MultiMessage.create();//解码后对象容器。list,可放多个消息
        do {
            Object obj = codec.decode(channel, buffer);//解码过程在DubboCodec类中的decode方法里
            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;
    }

    private void logMessageLength(Object result, int bytes) {
        if (bytes <= 0) {
            return;
        }
        if (result instanceof Request) {
            try {
                ((RpcInvocation) ((Request) result).getData()).setAttachment(
                        Constants.INPUT_KEY, String.valueOf(bytes));
            } catch (Throwable e) {
                /* ignore */
            }
        } else if (result instanceof Response) {
            try {
                ((RpcResult) ((Response) result).getResult()).setAttachment(
                        Constants.OUTPUT_KEY, String.valueOf(bytes));
            } catch (Throwable e) {
                /* ignore */
            }
        }
    }

}

分析DubboCodec类之前,先说下dubbo协议消息格式,它包括消息头和消息体:

前2个字节: 为协议魔数,固定值oxdabb

第三字节: 第1比特(0/1)表示是请求消息,还是响应消息 第2比特(0/1)表示是是否必须双向通信,即有请求,必有响应 第3比特(0/1)表示是是否是,心跳消息 第低5位比特,表示一个表示消息序列化的方式(1,是dubbo ,2,是hessian...)

第四字节: 只在响应消息中用到,表示响应消息的状态,是成功,失败等

第5-12字节: 8个字节,表示一个long型数字,是reqeustId

第13—16字节: 4个字节,表示消息体的长度(字节数)

消息体,不固定长度 是请求消息时,表示请求数据 是响应消息时,表示方法调用返回结果。

编码和解码主要是对消息头的设置和解析。序列化和反序列化主要是对消息体的操作。

先看DubboCodec的关系图:

DubboCodec类decode方法的实现在其父类ExchangeCodec中:

     //先看下类中定义的常量:
    // header length.消息头长度
    protected static final int HEADER_LENGTH = 16;
    // magic header.
    protected static final short MAGIC = (short) 0xdabb;//1101 1010 1011 1011魔数
    protected static final byte MAGIC_HIGH = Bytes.short2bytes(MAGIC)[0];//高字节
    protected static final byte MAGIC_LOW = Bytes.short2bytes(MAGIC)[1];//低字节
    // message flag.
    protected static final byte FLAG_REQUEST = (byte) 0x80;//1000,0000//表示消息类型
    protected static final byte FLAG_TWOWAY = (byte) 0x40;//0100,0000//表示是否双向通信
    protected static final byte FLAG_EVENT = (byte) 0x20;//0010,0000//表示是否是心跳事件
    protected static final int SERIALIZATION_MASK = 0x1f;//0001,1111/表示是序列化实现类型
  /***
     * 解码入口方法
     * @param channel
     * @param buffer
     * @return
     * @throws IOException
     */
    public Object decode(Channel channel, ChannelBuffer buffer) throws IOException {
        //取得可读的字节数
        int readable = buffer.readableBytes();
        //header 最大16字节
        byte[] header = new byte[Math.min(readable, HEADER_LENGTH)];
        //从buffer中读16字节到header数组中
        buffer.readBytes(header);
        //调用本身decode方法
        return decode(channel, buffer, readable, header);
    }

     /***
     * 具体协议解析方法,本方法主要是读取验证消息头的过程
     * @param channel
     * @param buffer
     * @param readable
     * @param header
     * @return
     * @throws IOException
     */
    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) {
            //前两字节,不是dubbo魔数
            int length = header.length;
            //如果可读的字节数目,大于16字节
            if (header.length < readable) {
                //给header扩容到readable大小
                header = Bytes.copyOf(header, readable);
                //把buffer剩下的字节读到header中,这里多于16字节
                buffer.readBytes(header, length, readable - length);
            }
            //上层方法,第一字节已经验证过,这个从第二字节开始验证。
            for (int i = 1; i < header.length - 1; i++) {
                //如果发现,后续字节有dubbo协议的开头
                if (header[i] == MAGIC_HIGH && header[i + 1] == MAGIC_LOW) {
                    //重置readerIndex 位置,到魔数开始的地方。
                    buffer.readerIndex(buffer.readerIndex() - header.length + i);
                    //把魔数开始位置,以前的数据,覆盖赋给header(下面调用super.decode解析)
                    header = Bytes.copyOf(header, i);
                    break;
                }
            }
            //上层方法是dubbo telnet 编解码实现
            return super.decode(channel, buffer, readable, header);
        }
        // check length.
        if (readable < HEADER_LENGTH) {//小于16字节,返回 需要更多对象
            return DecodeResult.NEED_MORE_INPUT;
        }

        //get data length.读取header[]最后四字节,构造一个int的数据,
        //根据dubbo协议,这个是消息体的长度
        int len = Bytes.bytes2int(header, 12);
        //检查数据大小
        checkPayload(channel, len);//默认为8M

        int tt = len + HEADER_LENGTH;//总的消息大小,消息头加消息实体
        if (readable < tt) {//如果可读取的,不够消息总大小,就返回 需要更多数据
            return DecodeResult.NEED_MORE_INPUT;
        }

        // limit input stream.这个时候,buffer的readerIndex位置已是,读完header后的位置,接下来的len长度的数据,全是消息体的数据。
        ChannelBufferInputStream is = new ChannelBufferInputStream(buffer, len);

        try {
            //解码反序列化成Reqeust或者Response对象,decodeBody方法被子类DubboCodec重写了
            //这里要看DubboCodec的decodeBody的方法
            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);
                }
            }
        }
    }

DubboCodec类的decodeBody方法:

/***
     * 解码,是从输出流 is 取字节数据,经反序列化,构造Request 和Response对象的过程。
     * @param channel
     * @param is
     * @param header
     * @return
     * @throws IOException
     */
    protected Object decodeBody(Channel channel, InputStream is, byte[] header) throws IOException {
        //消息头第3字节和SERIALIZATION_MASK&操作后,就可以得到,序列化/反序列化方案
        byte flag = header[2], proto = (byte) (flag & SERIALIZATION_MASK);
        //根据序列化方案id,或者url指定,通过spi机制去获取序列化实现。dubbo协议默认用hession2序列化方案
        //是放在消息头flag 里的。这里proto 值是2
        //获取具体用序列化/反序列化实现
        Serialization s = CodecSupport.getSerialization(channel.getUrl(), proto);
        // get request id.header字节从第5到12 8个字节,是请求id
        long id = Bytes.bytes2long(header, 4);
        if ((flag & FLAG_REQUEST) == 0) {//根据flag&FLAG_REQUEST后,判断需要解码的消息类型
            // decode response.
            Response res = new Response(id);
            if ((flag & FLAG_EVENT) != 0) {
                res.setEvent(Response.HEARTBEAT_EVENT);
            }
            // get status. 获取响应 状态 成功,失败等
            byte status = header[3];
            res.setStatus(status);
            if (status == Response.OK) {//返回结果状态 ok成功
                try {
                    Object data;
                    if (res.isHeartbeat()) {
                        data = decodeHeartbeatData(channel, deserialize(s, channel.getUrl(), is));
                    } else if (res.isEvent()) {//事件消息,反序列化
                        data = decodeEventData(channel, deserialize(s, channel.getUrl(), is));
                    } else {//业务调用结果消息 解码构造 DecodeableRpcResult 对象的过程
                        DecodeableRpcResult result;
                        if (channel.getUrl().getParameter(
                                Constants.DECODE_IN_IO_THREAD_KEY,
                                Constants.DEFAULT_DECODE_IN_IO_THREAD)) {//是否在io 线程内解码
                            result = new DecodeableRpcResult(channel, res, is,
                                    (Invocation) getRequestData(id), proto);
                            //!!!Response消息反序列化 就是把调用结果返回值 从is里反序列化出来,放在 DecodeableRpcResult类的result 字段的过程。
                            result.decode();
                        } else {
                            //不在io线程解码,要先通过readMessageData方法把调用结果数组取出后,
                            //放在UnsafeByteArrayInputStream对象,存在DecodeableRpcResult对象里,后续通过上层方法解码。
                            result = new DecodeableRpcResult(channel, res,
                                    new UnsafeByteArrayInputStream(readMessageData(is)),
                                    (Invocation) getRequestData(id), proto);
                        }
                        data = result;
                    }

                    //同时把DecodeableRpcResult对象放入Response result字段。
                    res.setResult(data);
                } catch (Throwable t) {
                    if (log.isWarnEnabled()) {
                        log.warn("Decode response failed: " + t.getMessage(), t);
                    }
                    //异常处理,设置status和异常信息
                    res.setStatus(Response.CLIENT_ERROR);
                    res.setErrorMessage(StringUtils.toString(t));
                }
            } else {
                //异常处理,设置异常信息
                res.setErrorMessage(deserialize(s, channel.getUrl(), is).readUTF());
            }
            return res;
        } else {//反解码Requset 消息类型
            // decode request.
            Request req = new Request(id);
            req.setVersion("2.0.0");
            req.setTwoWay((flag & FLAG_TWOWAY) != 0);
            if ((flag & FLAG_EVENT) != 0) {
                req.setEvent(Request.HEARTBEAT_EVENT);
            }
            try {
                Object data;
                if (req.isHeartbeat()) {
                    data = decodeHeartbeatData(channel, deserialize(s, channel.getUrl(), is));
                } else if (req.isEvent()) {
                    data = decodeEventData(channel, deserialize(s, channel.getUrl(), is));
                } else {//业务调用请求消息 解码构造 DecodeableRpcInvocation 对象的过程
                    DecodeableRpcInvocation inv;
                    if (channel.getUrl().getParameter(
                            Constants.DECODE_IN_IO_THREAD_KEY,
                            Constants.DEFAULT_DECODE_IN_IO_THREAD)) {//是否 io 线程内解码
                        inv = new DecodeableRpcInvocation(channel, req, is, proto);
                        //!!!Requset 类型反序列化方法
                        inv.decode();
                    } else {
                        inv = new DecodeableRpcInvocation(channel, req,
                                new UnsafeByteArrayInputStream(readMessageData(is)), proto);
                    }
                    data = inv;
                }
                //同时把DecodeableRpcInvocation对象放入Request data字段。
                req.setData(data);
            } catch (Throwable t) {
                if (log.isWarnEnabled()) {
                    log.warn("Decode request failed: " + t.getMessage(), t);
                }
                // bad request 异常请求对象设置
                req.setBroken(true);
                req.setData(t);
            }
            return req;
        }
    }

    /***
     * 获取反序列化方案
     * @param serialization
     * @param url
     * @param is
     * @return
     * @throws IOException
     */
    private ObjectInput deserialize(Serialization serialization, URL url, InputStream is)
            throws IOException {
        return serialization.deserialize(url, is);
    }

    /***
     * 读取is 里的可用数据
     * @param is
     * @return
     * @throws IOException
     */
    private byte[] readMessageData(InputStream is) throws IOException {
        if (is.available() > 0) {
            byte[] result = new byte[is.available()];
            is.read(result);
            return result;
        }
        return new byte[]{};
    }

RPC调用请求:DecodeableRpcInvocation 类反序列化方法:

public void decode() throws Exception {
        if (!hasDecoded && channel != null && inputStream != null) {
            try {//具体在decode重载方法里
                decode(channel, inputStream);
            } catch (Throwable e) {//异常请求设置
                if (log.isWarnEnabled()) {
                    log.warn("Decode rpc invocation failed: " + e.getMessage(), e);
                }
                request.setBroken(true);
                request.setData(e);
            } finally {
                hasDecoded = true;
            }
        }
    }

    /**
     * 反序列化,解码 通过反序列化还原
     * RpcInvocation 类的
     * private String methodName;
     private Class<?>[] parameterTypes;
     private Object[] arguments;
     private Map<String, String> attachments; 是个属性值,就像在客户端请求时设置的一样。
     * @param channel channel.
     * @param input   input stream.
     * @return
     * @throws IOException
     */
    public Object decode(Channel channel, InputStream input) throws IOException {
        //获取反序列化方案
        ObjectInput in = CodecSupport.getSerialization(channel.getUrl(), serializationType)
                .deserialize(channel.getUrl(), input);
        //反序列化出,dubbo版本,路径,服务版本信息,设置到attachments里,这读取顺序和序列化时的写入顺序也一致
        //当然,序列化方案也一致。这里默认都是hissen2
        //调用ObjectInput的readUTF()反序列化方法,依次获取调用信息
        setAttachment(Constants.DUBBO_VERSION_KEY, in.readUTF());
        setAttachment(Constants.PATH_KEY, in.readUTF());
        setAttachment(Constants.VERSION_KEY, in.readUTF());
        //读取方法名
        setMethodName(in.readUTF());
        //反序列化,方法请求参数类型,
        try {
            Object[] args;
            Class<?>[] pts;
            String desc = in.readUTF();
            if (desc.length() == 0) {
                pts = DubboCodec.EMPTY_CLASS_ARRAY;
                args = DubboCodec.EMPTY_OBJECT_ARRAY;
            } else {
                pts = ReflectUtils.desc2classArray(desc);
                args = new Object[pts.length];
                for (int i = 0; i < args.length; i++) {
                    try {
                        //更具类型读取请求参数值
                        //调用ObjectInput的readObject()反序列化方法,反序列化出参数值
                        args[i] = in.readObject(pts[i]);
                    } catch (Exception e) {
                        if (log.isWarnEnabled()) {
                            log.warn("Decode argument failed: " + e.getMessage(), e);
                        }
                    }
                }
            }
            //设置保存请求参数类型
            setParameterTypes(pts);
            //反序列化,attachment map
            //调用ObjectInput的readObject()反序列化方法,反序列化出attachemnet值
            Map<String, String> map = (Map<String, String>) in.readObject(Map.class);
            if (map != null && map.size() > 0) {
                Map<String, String> attachment = getAttachments();
                if (attachment == null) {
                    attachment = new HashMap<String, String>();
                }
                attachment.putAll(map);
                setAttachments(attachment);
            }
            //decode argument ,may be callback 回调参数设置,这个再说。
            for (int i = 0; i < args.length; i++) {
                args[i] = decodeInvocationArgument(channel, this, pts, i, args[i]);
            }
            //保存请求参数值
            setArguments(args);

        } catch (ClassNotFoundException e) {
            throw new IOException(StringUtils.toString("Read invocation data failed.", e));
        }
        return this;
    }

RPC调用结果:DecodeableRpcResult 反序列化 方法decode()

public void decode() throws Exception {
        if (!hasDecoded && channel != null && inputStream != null) {
            try {//具体实现在decode(channel, inputStream)重载方法里
                decode(channel, inputStream);
            } catch (Throwable e) {//设置异常返回response
                if (log.isWarnEnabled()) {
                    log.warn("Decode rpc result failed: " + e.getMessage(), e);
                }
                response.setStatus(Response.CLIENT_ERROR);
                response.setErrorMessage(StringUtils.toString(e));
            } finally {
                hasDecoded = true;
            }
        }
    }

     /***
     * 反序列化,解码过程,读取input的调用结果字节数据,经反序列化成方法返回类型对象
     * 并发放回结果设置到RpcResult的result字段里
     * 以及异常返回字段的设置。
     * @param channel channel.
     * @param input   input stream.
     * @return
     * @throws IOException
     */
    public Object decode(Channel channel, InputStream input) throws IOException {
        //获取反序列化方案
        ObjectInput in = CodecSupport.getSerialization(channel.getUrl(), serializationType)
                .deserialize(channel.getUrl(), input);
        //反序列化出 结果标识 调用ObjectInput的readByte方法,反序列化出一个byte值
        byte flag = in.readByte();//
        switch (flag) {
            case DubboCodec.RESPONSE_NULL_VALUE://null 值
                break;
            case DubboCodec.RESPONSE_VALUE: //非空值
                try {
                    //根据invocation获取调用方法的放回类型
                    Type[] returnType = RpcUtils.getReturnTypes(invocation);
                    //根据返回类型,反序列出结果并这是到RpcResult 的result字段里。
                    //void 类型 结果值 是null ;int 等基本类型,自动装箱 Integer;
                    //具体调用ObjectInput的readObject重载的两个方法,反序列出结果对象
                    setValue(returnType == null || returnType.length == 0 ? in.readObject() :
                            (returnType.length == 1 ? in.readObject((Class<?>) returnType[0])
                                    : in.readObject((Class<?>) returnType[0], returnType[1])));
                } catch (ClassNotFoundException e) {
                    throw new IOException(StringUtils.toString("Read response data failed.", e));
                }
                break;
            case DubboCodec.RESPONSE_WITH_EXCEPTION://异常信息反序列化,设置到exception字段
                try {
                   //具体调用ObjectInput的readObject重载的两个方法,反序列出异常对象
                    Object obj = in.readObject();
                    if (obj instanceof Throwable == false)
                        throw new IOException("Response data error, expect Throwable, but get " + obj);
                    setException((Throwable) obj);
                } catch (ClassNotFoundException e) {
                    throw new IOException(StringUtils.toString("Read response data failed.", e));
                }
                break;
            default:
                throw new IOException("Unknown result flag, expect '0' '1' '2', get " + flag);
        }
        return this;
    }

下一篇再说说反序列化。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏后端之路

Excel导入改善

最近由于上线的需要,对于excel导入导出的效率提出了要求。 那么一方面我们对于db落库的业务尽量走批量, 其次对于导入数据的校验以及对应成实体也做出了改善。...

5517
来自专栏JavaEE

做Java开发,你需要了解这些前言

在开发中,我们写的代码肯定是越少越好,代码层次越清晰越好。那么下面就介绍一些可以减少代码量、可以让结构更清晰的好东西。本文涉及vo、dto的使用、全局异常处理、...

1343
来自专栏分布式系统进阶

Kafka中Message存储相关类大揭密Kafka源码分析-汇总

1691
来自专栏美码师

补习系列-springboot 参数校验详解

在定义 Restful 风格的接口时,通常会采用 PathVariable 指定关键业务参数,如下:

5611
来自专栏恰同学骚年

Hadoop学习笔记—7.计数器与自定义计数器

  在上图所示中,计数器有19个,分为四个组:File Output Format Counters、FileSystemCounters、File Input...

1022
来自专栏iOS122-移动混合开发研究院

【自问自答】关于 Swift 的几个疑问

感觉自己给自己释疑,也是一个极为有趣的过程。这次,我还新增了“猜想”一栏,来尝试回答一些暂时没有足够资料支撑的问题。 Swift 版本是:4.0.3。不同版本的...

3116
来自专栏后端沉思录

SpringMVC启动加载、请求分析

DispatcherServlet其实是一个Servlet,用于初始化各个功能的实现类,比如异常处理、视图处理、请求映射等;且继承了FrameworkServl...

1013
来自专栏源码之家

word如何自动分割成多个文档

3785
来自专栏冰霜之地

高效的序列化/反序列化数据方式 Protobuf

上篇文章中其实已经讲过了 encode 的过程,这篇文章以 golang 为例,从代码实现的层面讲讲序列化和反序列化的过程。

4845
来自专栏农夫安全

注入学习之sqli-labs-6(第五次)

前言 上一次课讲解的是sql基于布尔型盲注,紧接着这节讲基于时间的盲注 布尔型盲注,是在我们判断网站是否存在注入的时候,网页不会暴漏错误信息,但会返回正确的页面...

3716

扫码关注云+社区