Thrift协议的服务模型-白红宇

一、Thrift介绍

Thrift是一个软件框架，用来进行可扩展且跨语言的服务的开发。它结合了功能强大的软件堆栈和代码生成引擎。其允许你定义一个简单的定义文件中的数据类型和服务接口。以作为输入文件，编译器生成代码用来方便地生成RPC客户端和服务器通信的无缝跨编程语言。

二、Thrift基础架构

Thrift 支持的数据类型

1、基本类型

　　bool: 布尔值

　　byte: 8位有符号整数

　　i16: 16位有符号整数

　　i32: 32位有符号整数

　　i64: 64位有符号整数

　　double: 64位浮点数

　　string: UTF-8编码的字符串

　　binary: 二进制串

2、结构体类型

　　struct: 定义的结构体对象

3、容器类型

Thrift容器与类型密切相关，它与当前流行编程语言提供的容器类型相对应，采用java泛型风格表示。Thrift提供了3种容器类型：

　　list<t1>: 一系列t1类型的元素组成的有序表，元素可以重复

　　set: <t1>：一系列t1类型的元素组成的无序表，元素唯一

　　map<t1,t2>：key/value对（key的类型是t1且唯一，value类型是t2）

容器中的元素类型可以是除了service意外的任何合法thrift类型（包括结构体和异常）

4、异常类型：

　　exception: 异常类型

5、服务类型：

　　service: 具体对应服务的类

协议

Thrift可以让你选择客户端与服务端之间传输通信协议的类别，在传输协议上总体上划分为文本(text)和二进制(binary)传输协议, 为节约带宽，提供传输效率，一般情况下使用二进制类型的传输协议为多数，但有时会还是会使用基于文本类型的协议，这需要根据项目/产品中的实际需求：

1、TBinaryProtocol – 二进制编码格式进行数据传输。

2、TCompactProtocol – 这种协议非常有效的，使用Variable-Length Quantity (VLQ) 编码对数据进行压缩。

3、TJSONProtocol – 使用JSON的数据编码协议进行数据传输。

4、TSimpleJSONProtocol – 这种节约只提供JSON只写的协议，适用于通过脚本语言解析

5、TDebugProtocol – 在开发的过程中帮助开发人员调试用的，以文本的形式展现方便阅读。

传输层

1、TSocket- 使用堵塞式I/O进行传输，也是最常见的模式。

2、TFramedTransport- 使用非阻塞方式，按块的大小，进行传输，类似于Java中的NIO。

3、TFileTransport- 顾名思义按照文件的方式进程传输，虽然这种方式不提供Java的实现，但是实现起来非常简单。

4、TMemoryTransport- 使用内存I/O，就好比Java中的ByteArrayOutputStream实现。

5、TZlibTransport- 使用执行zlib压缩，不提供Java的实现。

三、Thrift网络服务模型

Thrif 提供网络模型：单线程、多线程、事件驱动。从另一个角度划分为：阻塞服务模型、非阻塞服务模型。

阻塞服务

TSimpleServer：简单的单线程服务模型，主要用于测试

TThreadPoolServer：多线程服务模型，使用标准的阻塞式IO，预先创建一组线程处理请求

非阻塞服务模型

TNonblockingServer：多线程服务模型，使用非阻塞式IO（需使用TFramedTransport数据传输方式）,只有一个线程来处理消

THsHaServer：半同步半异步的服务模型，一个单独的线程用来处理网络I/O，一个worker线程池用来进行消息的处理

TThreadedSelectorServer：允许你用多个线程来处理网络I/O。它维护了两个线程池，一个用来处理网络I/O，另一个用来进行请求的处理

1、TSimpleServer

TSimpleServer实现是非常的简单，循环监听新请求的到来并完成对请求的处理，是个单线程阻塞模型。由于是一次只能接收和处理一个socket连接，效率比较低，在实际开发过程中很少用到它。

/** * 注册服务端 *  单线程服务模型，使用标准的阻塞式IO，只有一个线程处理请求 */public static void startSimpleServer(AdditionService.Processor
     
       processor) {    try {        TServerTransport serverTransport = new TServerSocket(9090);// 设置服务端口        // 单线程服务模型        TSimpleServer.Args tArgs = new TSimpleServer.Args(serverTransport);        tArgs.processor(processor);        // 客户端协议要一致        tArgs.protocolFactory(new TBinaryProtocol.Factory());        TServer server = new TSimpleServer(tArgs);        System.out.println("Starting the simple server...");        server.serve();    } catch (Exception e) {        e.printStackTrace();    }}

2、TThreadPoolServer

ThreadPoolServer为解决了TSimpleServer不支持并发和多连接的问题, 引入了线程池。但仍然是多线程阻塞模式即实现的模型是One Thread Per Connection。

线程池采用能线程数可伸缩的模式，线程池中的队列采用同步队列(SynchronousQueue)。

ThreadPoolServer拆分了监听线程(accept)和处理客户端连接的工作线程(worker), 监听线程每接到一个客户端, 就投给线程池去处理。

/** * 注册服务端 *     线程池服务模型，使用标准的阻塞式IO，预先创建一组线程处理请求 */public static void startMultipleServer(AdditionService.Processor
     
       processor) {    try {        TServerTransport serverTransport = new TServerSocket(9090);// 设置服务端口        //线程池服务模型，使用标准的阻塞式IO，预先创建一组线程处理请求。        TThreadPoolServer.Args tArgs = new TThreadPoolServer.Args(serverTransport);        tArgs.processor(processor);        // 客户端协议要一致        tArgs.protocolFactory(new TBinaryProtocol.Factory());        TServer server = new TThreadPoolServer(tArgs);        System.out.println("Hello TThreadPoolServer....");        server.serve(); // 启动服务    } catch (Exception e) {        e.printStackTrace();    }}

这两个服务端可以处理同样的客户端：

/** * 客户端调用 * 阻塞 * Created by Administrator on 2017/1/12. */public class AdditionClient {    public static void main(String[] args) {        try {            TTransport transport;            // 设置传输通道            transport = new TSocket("localhost", 9090);//使用堵塞式I/O进行传输            transport.open();            // 协议要和服务端一致            // 使用二进制协议            TProtocol protocol = new TBinaryProtocol(transport);            AdditionService.Client client = new AdditionService.Client(protocol);            System.out.println(client.add(100, 200));            transport.close();        } catch (TTransportException e) {            e.printStackTrace();        } catch (TException x) {            x.printStackTrace();        }    }}

TThreadPoolServer模式优点：

线程池模式中，数据读取和业务处理都交由线程池完成，主线程只负责监听新连接，因此在并发量较大时新连接也能够被及时接受。线程池模式比较适合服务器端能预知最多有多少个客户端并发的情况，这时每个请求都能被业务线程池及时处理，性能也非常高。

TThreadPoolServer模式缺点：

线程池模式的处理能力受限于线程池的工作能力，当并发请求数大于线程池中的线程数时，新请求也只能排队等待。

3、TNonblockingServer

TNonblockingServer采用单线程非阻塞(NIO)的模式, 借助Channel/Selector机制, 采用IO事件模型来处理。所有的socket都被注册到selector中，在一个线程中通过seletor循环监控所有的socket，每次selector结束时，处理所有的处于就绪状态的socket，对于有数据到来的socket进行数据读取操作，对于有数据发送的socket则进行数据发送，对于监听socket则产生一个新业务socket并将其注册到selector中。

/** * TNonblockingServer采用单线程非阻塞(NIO)的模式 * @param processor */public static void nonBlockingServer(AdditionService.Processor
     
       processor) {    try {        // 传输通道 - 非阻塞方式        TNonblockingServerSocket serverTransport = new TNonblockingServerSocket(9090);// 设置服务端口        // 异步IO，需要使用TFramedTransport，它将分块缓存读取。        TNonblockingServer.Args tArgs = new TNonblockingServer.Args(serverTransport);        tArgs.processor(processor);        tArgs.transportFactory(new TFramedTransport.Factory());        // 使用高密度二进制协议        tArgs.protocolFactory(new TCompactProtocol.Factory());        // 使用非阻塞式IO，服务端和客户端需要指定TFramedTransport数据传输的方式        TServer server = new TNonblockingServer(tArgs);        System.out.println("Starting the simple server...");        server.serve();// 启动服务    } catch (Exception e) {        e.printStackTrace();    }}

查看TNonblockingServer的源码，我们可以看到select（）方法：

private void select() {    try {        this.selector.select();        Iterator e = this.selector.selectedKeys().iterator();        while(!TNonblockingServer.this.stopped_ && e.hasNext()) {            SelectionKey key = (SelectionKey)e.next();            e.remove();            if(!key.isValid()) {                this.cleanupSelectionKey(key);            } else if(key.isAcceptable()) {                this.handleAccept();            } else if(key.isReadable()) {                this.handleRead(key);            } else if(key.isWritable()) {                this.handleWrite(key);            } else {                TNonblockingServer.this.LOGGER.warn("Unexpected state in select! " + key.interestOps());            }        }    } catch (IOException var3) {        TNonblockingServer.this.LOGGER.warn("Got an IOException while selecting!", var3);    }}

select代码里对accept/read/write等IO事件进行监控和处理, 但由于这是单线程处理，所以当遇到handler里有阻塞的操作时, 会导致整个服务被阻塞住。

TNonblockingServer模式优点：

相比于TSimpleServer效率提升主要体现在IO多路复用上，TNonblockingServer采用非阻塞IO，同时监控多个socket的状态变化；

TNonblockingServer模式缺点：

TNonblockingServer模式在业务处理上还是采用单线程顺序来完成，在业务处理比较复杂、耗时的时候，例如某些接口函数需要读取数据库执行时间较长，此时该模式效率也不高，因为多个调用请求任务依然是顺序一个接一个执行。

4、THsHaServer

THsHaServer类是TNonblockingServer类的子类，为解决TNonblockingServer的缺点, THsHa引入了线程池去处理, 其模型把读写任务放到线程池去处理即多线程非阻塞模式。HsHa是: Half-sync/Half-async的处理模式, Half-aysnc是在处理IO事件上(accept/read/write io), Half-sync用于handler对rpc的同步处理上。因此可以认为THsHaServer半同步半异步。

/** *  THsHaServer *  THsHaServer类是TNonblockingServer类的子类 * @param processor */public static void startTHsHaServer(AdditionService.Processor
     
       processor) {    try {        // 传输通道 - 非阻塞方式        TNonblockingServerSocket serverTransport = new TNonblockingServerSocket(9090);// 设置服务端口        // 异步IO，需要使用TFramedTransport，它将分块缓存读取。        THsHaServer.Args tArgs = new THsHaServer.Args(serverTransport);        tArgs.processor(processor);        tArgs.transportFactory(new TFramedTransport.Factory());        // 使用高密度二进制协议        tArgs.protocolFactory(new TCompactProtocol.Factory());        // 使用非阻塞式IO，服务端和客户端需要指定TFramedTransport数据传输的方式        TServer server = new THsHaServer(tArgs);        System.out.println("Starting the simple server...");        server.serve();// 启动服务    } catch (Exception e) {        e.printStackTrace();    }}

THsHaServer的优点：

与TNonblockingServer模式相比，THsHaServer在完成数据读取之后，将业务处理过程交由一个线程池来完成，主线程直接返回进行下一次循环操作，效率大大提升；

THsHaServer的缺点：

主线程需要完成对所有socket的监听以及数据读写的工作，当并发请求数较大时，且发送数据量较多时，监听socket上新连接请求不能被及时接受。

5、TThreadedSelectorServer

TThreadedSelectorServer是大家广泛采用的服务模型，其多线程服务器端使用非堵塞式I/O模型，是对TNonblockingServer的扩充, 其分离了Accept和Read/Write的Selector线程, 同时引入Worker工作线程池。

（1）一个AcceptThread线程对象，专门用于处理监听socket上的新连接；

（2）若干个SelectorThread对象专门用于处理业务socket的网络I/O操作，所有网络数据的读写均是有这些线程来完成；

（3）一个负载均衡器SelectorThreadLoadBalancer对象，主要用于AcceptThread线程接收到一个新socket连接请求时，决定将这个新连接请求分配给哪个SelectorThread线程。

（4）一个ExecutorService类型的工作线程池，在SelectorThread线程中，监听到有业务socket中有调用请求过来，则将请求读取之后，交给ExecutorService线程池中的线程完成此次调用的具体执行

MainReactor就是Accept线程, 用于监听客户端连接, SubReactor采用IO事件线程(多个), 主要负责对所有客户端的IO读写事件进行处理. 而Worker工作线程主要用于处理每个rpc请求的handler回调处理(这部分是同步的)。因此其也是Half-Sync/Half-Async（半异步-半同步）的。

TThreadedSelectorServer模式对于大部分应用场景性能都不会差，因为其有一个专门的线程AcceptThread用于处理新连接请求，因此能够及时响应大量并发连接请求；另外它将网络I/O操作分散到多个SelectorThread线程中来完成，因此能够快速对网络I/O进行读写操作，能够很好地应对网络I/O较多的情况。

/** *  TThreadedSelectorServer *  是多线程服务器端使用非堵塞式I/O模型 * @param processor */    public static void startTThreadedSelectorServer(AdditionService.Processor
     
       processor) {        try {            // 传输通道 - 非阻塞方式            TNonblockingServerSocket serverTransport = new TNonblockingServerSocket(9090);// 设置服务端口            // 异步IO，需要使用TFramedTransport，它将分块缓存读取。            TThreadedSelectorServer.Args tArgs = new TThreadedSelectorServer.Args(serverTransport);            tArgs.processor(processor);            tArgs.transportFactory(new TFramedTransport.Factory());            // 使用高密度二进制协议            tArgs.protocolFactory(new TCompactProtocol.Factory());            // 使用非阻塞式IO，服务端和客户端需要指定TFramedTransport数据传输的方式            TServer server = new TThreadedSelectorServer(tArgs);            System.out.println("Starting the simple server...");            server.serve();// 启动服务        } catch (Exception e) {            e.printStackTrace();        }    }

上面三种非阻塞服务模型可以处理这样的客户端：

public static void main(String[] args) {    try {        // 设置传输通道，对于非阻塞服务，需要使用TFramedTransport，它将数据分块发送        TSocket socket = new TSocket("localhost", 9090);//使用堵塞式I/O进行传输        //使用非阻塞方式，按块的大小，进行传输，类似于Java中的NIO        TTransport transport = new TFramedTransport(socket);        // 协议要和服务端一致        // 使用高密度二进制协议        TProtocol protocol = new TCompactProtocol(transport);        AdditionService.Client client = new AdditionService.Client(protocol);        transport.open();        System.out.println(client.add(100, 100));        transport.close();    } catch (TTransportException e) {        e.printStackTrace();    } catch (TException x) {        x.printStackTrace();    }}

由于能力有限，文章有很多不足之处，请大家不吝赐教，剩下的代码我会上传源码，大家可以下载下来查看

参考：