rpc框架之 thrift 学习 2 - 基本概念

thrift的基本构架:

上图源自:http://jnb.ociweb.com/jnb/jnbJun2009.html

底层Underlying I/O以上的部分,都是由thrift编译器生成的代码,其中:

Your Code 这是根据thrift文件中定义的dto及service接口方法

FooService.Client及FooService.Processer是thrift生成的用于客户端及服务端的标准代码

Foo.read/write 参数对象及结果对象在传输时,最终需要在client、server间进行重写,红色框指的就是这个

TProtocal 指传输的内容是啥?(二进制?Json ? )由于TProtocal是一个抽象类,因此最终调用时,如果想从BinaryProtocal换成JsonProtocal,这部分代码也不用重新生成

TTransport 指用什么方式传输?(Scoket? Memory?File?)同样,TTransport是抽象类,运行时由具体子类决定运输方式

最底层的Underlying I/O则是依赖于各种语言的实现,负责底层的网络通讯,thrift最初是由c++写的,理论上讲,c++上的性能应该最好。

上一章的demo为例如,QueryParameter及DemoService的类图如下:

点击图片可以查看大图,从类图上看,大量使用了内部类(inner class),对于dto对象,内部类基本上分为_Fileds, Schema,SchemaFactory 三类,

_Fields 是一个枚举,罗列了dto的各种属性成员,

Schema 封装了write/read方法

SchemaFactory 是一个工厂,用于创建Schema实例

服务接口的类图,就有点复杂了,密密麻麻象蜘蛛网,除了刚才的三大类外,DemoService的Inner Class中还有Client、Processor等类,大家有兴趣可以慢慢看。

TProtocal : 传输的内容(即:What? )

从类图上看,支持 压缩格式、二进制格式、Json格式 等。

TTransport : 传输的方式(即:How? ) 

Thrift支持的传输方式非常多,从类的命名就能大概看出一二。

TServer: Server的类图如下

基本上分为二大类:一类是同步阻塞的Server,一类是非阻塞模式的Server,其中THsHaServer是一个Half-Sync/Half-Async 半同步,半异步的server

meta_data 元数据

类图中的xxxMetaData,基本对应了 列表、K-V映射、(无重复元素)集合、结构(即:类)、枚举以及字段的元数据信息。

Schema :对不同类型的TProtocal的读写操作,在这里抽象出来。

Variable-Length Quantity VLQ 变长编码: Thirft采用TCompactProtocol序列化时之所以高效,跟VLQ变长编码有很大关系,直接借下面这张图来说吧:

整数106903,在java中我们知道int占用4个bytes,也就是32bit,高位字节如果不满,用0填充(最高位符号位除外), 这样的话,很多用0填充的高位字节位置其实是浪费的,VLQ的基本思路是将2进制每7位分组,这样106903的2进制就可以分成3组,然后每1组的最高位设为1或0,如果为1,表示相邻的下一个字节还有内容,要继续读取,如果该位置为0,则表示结束了。

这样的话,106903最终只需要3个字节就可以存储了,节省了1个字节。

上述这一堆概念在运行时,是如何串起来的呢?

可以从TServer的部分源码中略知一二:

public abstract class TServer {

  public static class Args extends AbstractServerArgs<Args> {
    public Args(TServerTransport transport) {
      super(transport);
    }
  }

  public static abstract class AbstractServerArgs<T extends AbstractServerArgs<T>> {
    final TServerTransport serverTransport;
    TProcessorFactory processorFactory;
    TTransportFactory inputTransportFactory = new TTransportFactory();
    TTransportFactory outputTransportFactory = new TTransportFactory();
    TProtocolFactory inputProtocolFactory = new TBinaryProtocol.Factory();
    TProtocolFactory outputProtocolFactory = new TBinaryProtocol.Factory();

    public AbstractServerArgs(TServerTransport transport) {
      serverTransport = transport;
    }

    ...

  /**
   * Core processor
   */
  protected TProcessorFactory processorFactory_;

  /**
   * Server transport
   */
  protected TServerTransport serverTransport_;

  /**
   * Input Transport Factory
   */
  protected TTransportFactory inputTransportFactory_;

  /**
   * Output Transport Factory
   */
  protected TTransportFactory outputTransportFactory_;

  /**
   * Input Protocol Factory
   */
  protected TProtocolFactory inputProtocolFactory_;

  /**
   * Output Protocol Factory
   */
  protected TProtocolFactory outputProtocolFactory_;

  private boolean isServing;

  protected TServerEventHandler eventHandler_;

  // Flag for stopping the server
  // Please see THRIFT-1795 for the usage of this flag
  protected volatile boolean stopped_ = false;

  protected TServer(AbstractServerArgs args) {
    processorFactory_ = args.processorFactory;
    serverTransport_ = args.serverTransport;
    inputTransportFactory_ = args.inputTransportFactory;
    outputTransportFactory_ = args.outputTransportFactory;
    inputProtocolFactory_ = args.inputProtocolFactory;
    outputProtocolFactory_ = args.outputProtocolFactory;
  }

从上述源码可以看出,TServer 中包含有input/output二类TProtocol,即:体现了 数据进来和出去时传输的格式(Binary? Json?...),另外还有input与output的Transport,即:数据进来和出去的时候,如何传输?(Scoket? File?...),另外还有一个Processor,其子类是通过IDL(thrift定义文件)生成的,运行时必须传递进来具体的子类。

这样,传递什么数据(what)?用什么方式传输(how)? 以及数据如何处理(process)?都有了

而且从源码中,可以发现默认的input/output Protocol都是BinaryProtocol(14,15行)。

Server端启动的时序图如下:

Client调用的时序图:

注:上面这二张序列图均出自https://www.ibm.com/developerworks/cn/java/j-lo-apachethrift/

有了这些整体的概念后,上一篇中的示例,如果我们想换成TCompactProtocal,Client与Server的代码都得同步修改,这样二边才能一致:

Client端:

public class ThriftClient {

    public static void main(String[] args) {

        try {
            TTransport transport;
            transport = new TSocket("localhost", 9090);
            transport.open();

            //TProtocol protocol = new TBinaryProtocol(transport);
            TProtocol protocol = new TCompactProtocol(transport);
            DemoService.Client client = new DemoService.Client(protocol);
...

Server端:

    public static void simple(DemoService.Processor processor) {
        try {
            TServerTransport serverTransport = new TServerSocket(9090);
            Args args = new Args(serverTransport);
            args.outputProtocolFactory(new TCompactProtocol.Factory());
            args.inputProtocolFactory(new TCompactProtocol.Factory());
            TServer server = new TSimpleServer(args.processor(processor));
            //TServer server = new TSimpleServer(new Args(serverTransport).processor(processor));
            System.out.println("Starting the simple server...");
            server.serve();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

最后,对TCompactProtocol、TBinaryProtocol、TJSONProtocol这三种序列化协议简单测试下效率:

    @Test
    public void testProtocol() throws Exception {
        QueryParameter parameter = getQueryParameter();

        //CompactProtocol测试
        TSerializer serializerCompact = new TSerializer(new TCompactProtocol.Factory());
        byte[] bytes1 = serializerCompact.serialize(parameter);
        System.out.println("CompactProtocol序列后的byte数组长度:" + bytes1.length);

        //BinaryProtocol测试
        TSerializer serializerBinary = new TSerializer(new TBinaryProtocol.Factory());
        byte[] bytes2 = serializerBinary.serialize(parameter);
        System.out.println("BinaryProtocol序列后的byte数组长度:" + bytes2.length);

        //JsonProtocol测试
        TSerializer serializerJson = new TSerializer(new TJSONProtocol.Factory());
        byte[] bytes3 = serializerJson.serialize(parameter);
        System.out.println("JsonProtocol序列后的byte数组长度:" + bytes3.length);
        System.out.println(serializerJson.toString(parameter));


        System.out.println("-----------");
        TDeserializer deserializerCompact = new TDeserializer(new TCompactProtocol.Factory());
        QueryParameter query1 = new QueryParameter();
        deserializerCompact.deserialize(query1, bytes1);
        System.out.println("CompactProtocol反序列化结果:" + query1.equals(parameter));

        TDeserializer deserializerBinary = new TDeserializer(new TBinaryProtocol.Factory());
        QueryParameter query2 = new QueryParameter();
        deserializerBinary.deserialize(query2, bytes2);
        System.out.println("BinaryProtocol反序列化结果:" + query2.equals(parameter));

        TDeserializer deserializerJSON = new TDeserializer(new TJSONProtocol.Factory());
        QueryParameter query3 = new QueryParameter();
        deserializerJSON.deserialize(query3, bytes3);
        System.out.println("JSONProtocol反序列化结果:" + query2.equals(parameter));
    }

    private QueryParameter getQueryParameter(){
        QueryParameter query = new QueryParameter();
        short start = 1;
        short end = 5;
        query.setAgeStart(start);
        query.setAgeEnd(end);
        return query;
    }

输出结果:

CompactProtocol序列后的byte数组长度:5 BinaryProtocol序列后的byte数组长度:11 JsonProtocol序列后的byte数组长度:29 {"1":{"i16":1},"2":{"i16":5}} ----------- CompactProtocol反序列化结果:true BinaryProtocol反序列化结果:true JSONProtocol反序列化结果:true

TCompactProtocol优势明显,序列后的bytes长度只有JSON的1/5左右,可以大幅减少网络传输量。

参考文章:

http://dongxicheng.org/search-engine/thrift-rpc/

http://blog.chinaunix.net/uid-20357359-id-2876170.html

http://jnb.ociweb.com/jnb/jnbJun2009.html

https://www.ibm.com/developerworks/cn/java/j-lo-apachethrift/

http://www.cnblogs.com/brucewoo/tag/Thrift/

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

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏mini188

java中的锁

java中有哪些锁 这个问题在我看了一遍<java并发编程>后尽然无法回答,说明自己对于锁的概念了解的不够。于是再次翻看了一下书里的内容,突然有点打开脑门的感觉...

4319
来自专栏Android 研究

OKHttp源码解析(五)--OKIO简介及FileSystem

okio是由square公司开发的,它补充了java.io和java.nio的不足,以便能够更加方便,快速的访问、存储和处理你的数据。OKHttp底层也是用该库...

2012
来自专栏Spark生态圈

[Spark SQL] 源码解析之Analyzer

Analyzer模块将Unresolved LogicalPlan结合元数据catalog进行绑定,最终转化为Resolved LogicalPlan。跟着代码...

1272
来自专栏祝威廉

ElasticSearch Aggregations GroupBy 实现源码分析

也就是按newtype 字段进行group by,然后对num求平均值。在我们实际的业务系统中,这种统计需求也是最多的。

3893
来自专栏SDNLAB

POF技术分享(三):Packet处理流程

前言: 之前对POF基本原理、POF交换机源码结构进行解读,但是,要想完成POF交换机的二次开发和拓展,有必要对POF交换机特有的数据包处理流程、POF交换机和...

37712
来自专栏jeremy的技术点滴

JVM的Finalization Delay引起的OOM

3948
来自专栏冰霜之地

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

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

5505
来自专栏wannshan(javaer,RPC)

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

由于rpc底层涉及网络编程接口,线程模型,网络数据结构,服务协议,细到字节的处理。牵涉内容较多,今天就先从一个点说起。 说说,dubbo通过netty框架做传...

5026
来自专栏恰童鞋骚年

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

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

1092
来自专栏ImportSource

Junit 5新特性全集

本文略长,但都是大白话,如果你能一口气看完,你赢了。 如果你来不及看这么长,那么建议你滑到文末,直接看黑体部分就知道大概了。 在5中的一个测试类的基本生命周期是...

48612

扫码关注云+社区

领取腾讯云代金券