JAVA RPC (六) 之thrift反序列化RPC消息体

我们来看一下服务端的简单实现，直接上thrift代码，很直观的来看一看thrift的server到底干了些什么

 1  public boolean process(TProtocol in, TProtocol out) throws TException {
 2         TMessage msg = in.readMessageBegin();
 3         ProcessFunction fn = (ProcessFunction)this.processMap.get(msg.name);
 4         if (fn == null) {
 5             TProtocolUtil.skip(in, (byte)12);
 6             in.readMessageEnd();
 7             TApplicationException x = new TApplicationException(1, "Invalid method name: '" + msg.name + "'");
 8             out.writeMessageBegin(new TMessage(msg.name, (byte)3, msg.seqid));
 9             x.write(out);
10             out.writeMessageEnd();
11             out.getTransport().flush();
12             return true;
13         } else {
14             fn.process(msg.seqid, in, out, this.iface);
15             return true;
16         }
17     }

TMessage msg = in.readMessageBegin();这段代码的意思是从client端获取到二进制数据时候，获取Tmessage实例，和client的wirteMessage方法进行对应，聪明的小伙伴这个时候一定会想到这个是怎么解析的
为什么没有解析消息体长度和消息体内容的部分代码，不要着急我们一步一步渗透，readMessageBegin核心代码如下

 1  public TMessage readMessageBegin() throws TException {
 2         int size = this.readI32();
 3         if (size < 0) {
 4             int version = size & -65536;
 5             if (version != -2147418112) {
 6                 throw new TProtocolException(4, "Bad version in readMessageBegin");
 7             } else {
 8                 return new TMessage(this.readString(), (byte)(size & 255), this.readI32());
 9             }
10         } else if (this.strictRead_) {
11             throw new TProtocolException(4, "Missing version in readMessageBegin, old client?");
12         } else {
13             return new TMessage(this.readStringBody(size), this.readByte(), this.readI32());
14         }
15     }

int size = this.readI32();是做了些什么，接着往下看

 1 public int readI32() throws TException {
 2         byte[] buf = this.i32rd;
 3         int off = 0;
 4         if (this.trans_.getBytesRemainingInBuffer() >= 4) {
 5             buf = this.trans_.getBuffer();
 6             off = this.trans_.getBufferPosition();
 7             this.trans_.consumeBuffer(4);
 8         } else {
 9             this.readAll(this.i32rd, 0, 4);
10         }
11 
12         return (buf[off] & 255) << 24 | (buf[off + 1] & 255) << 16 | (buf[off + 2] & 255) << 8 | buf[off + 3] & 255;
13     }

这个地方代码就很清晰了，原来是从trans里面或取得字节流数据。

1 private int readAll(byte[] buf, int off, int len) throws TException {
2         this.checkReadLength(len);
3         return this.trans_.readAll(buf, off, len);
4     }

 

继续能翻到最终的代码解析部分

private void readFrame() throws TTransportException {
        this.transport_.readAll(this.i32buf, 0, 4);
        int size = decodeFrameSize(this.i32buf);
        if (size < 0) {
            throw new TTransportException("Read a negative frame size (" + size + ")!");
        } else if (size > this.maxLength_) {
            throw new TTransportException("Frame size (" + size + ") larger than max length (" + this.maxLength_ + ")!");
        } else {
            byte[] buff = new byte[size];
            this.transport_.readAll(buff, 0, size);
            this.readBuffer_.reset(buff);
        }
    }

这里就很明显了，首先read一个四个字节的长度，这个int类型代表着后面要跟着接收的消息体的长度，来源是client的发送内容，然后在根据消息体长度在读出来所有的内容，然后缓存到readBuffer_中，然后再读取内容的时候直接从readBuffer_中获取字节流就可以了。

好了，服务端接收二进制数据解析的关键点就在这里了,接下来开始，我将用大量的篇幅开始讲解我的企业级RPC框架，满满的干货，感兴趣的小伙伴去我码云给个star吧！！！

https://gitee.com/a1234567891/koalas-rpc

koalas-RPC 个人作品，提供大家交流学习，有意见请私信，欢迎拍砖。客户端采用thrift协议，服务端支持netty和thrift的TThreadedSelectorServer半同步半异步线程模型，支持动态扩容，服务上下线，权重动态，可用性配置，页面流量统计等，持续为个人以及中小型公司提供可靠的RPC框架技术方案

更多学习内容请加高级java QQ群:825199617