博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
Thrift 简单使用
阅读量:4048 次
发布时间:2019-05-25

本文共 15151 字,大约阅读时间需要 50 分钟。

一.Thrift介绍

    thrift是一个软件框架,用来进行可扩展且跨语言的服务的开发。它结合了功能强大的软件堆栈和代码生成引擎,以构建在 C++, Java, Python, PHP, Ruby, Erlang, Perl, Haskell, C#, Cocoa, JavaScript, Node.js, Smalltalk, and OCaml 这些编程语言间无缝结合的、高效的服务。

    thrift最初由facebook开发用做系统内个语言之间的RPC通信 。

 2007年由facebook贡献到apache基金 ,08年5月进入apache孵化器 。
 支持多种语言之间的RPC(Remote Procedure Call Protocol 远程过程调用协议)方式的通信:其它语言client可以构造一个对象,调用相应的服务方法来调用java语言的服务 ,跨越语言的C/S RPC调用 。——来自百度

 Thrift在接口定义文档、数据类型支持、跨语言等方面存在优势,然而对服务地址和服务管理方面支持力度不够,并且存在类型侵入问题。——《搜狗商业平台Java技术实践》

    协议、传输层等其它细节可以参考 http://www.ibm.com/developerworks/cn/java/j-lo-apachethrift/

    关于protobuf和thrift的对比可以自行搜索,protobuf是谷歌出的二进制解析协议就好像JSON一样。

 

二.为什么选用Thrift

    1.处理类似的问题,我们会先想到WebService,不管是基于XML的jax-ws,还是基于JSON的jax-rs(按照表述性状态转移(REST)架构风格创建Web服务)都是基于Http协议的,Thrift可以基于TCP,效率上有优势;

    2.Thrift有两种常用的协议TCompactProtocol,TBinaryProtocol(默认),体积超小,使用起来比较麻烦,不如前两者轻便,但是对于高并发、数据传输量大、多语言环境比XML和JSON传输对象体积更小,处理效率更高。

 

三.IDL定义

 

Thrift 脚本可定义的数据类型包括以下几种类型:

 

1.基本类型:

    bool:布尔值,true 或 false,对应 Java 的 boolean

    byte:8 位有符号整数,对应 Java 的 byte

    i16:16 位有符号整数,对应 Java 的 short

    i32:32 位有符号整数,对应 Java 的 int

    i64:64 位有符号整数,对应 Java 的 long

    double:64 位浮点数,对应 Java 的 double

    string:未知编码文本或二进制字符串,对应 Java 的 String

 

2.结构体类型:

    struct:定义公共的对象,类似于 C 语言中的结构体定义,在 Java 中是一个 JavaBean

 

3.容器类型:

    list:对应 Java 的 ArrayList

    set:对应 Java 的 HashSet

    map:对应 Java 的 HashMap

 

4.异常类型:

    exception:对应 Java 的 Exception

 

5.服务类型:

    service:对应服务的类

 

四.Thrift生成DEMO

1.下载thrift

下面的地址下载thrift的可执行文件,目前的版本是thrift-0.9.2.exe

 

 

http://thrift.apache.org/download
 

 

2.定义IDL

定义两个IDL(.thrift)文件,ThriftStruct.thrift是Domain定义,ThriftService.thrift是Service方法的定义

注意Thrift服务不支持方法重载

 

 

namespace java com.gqshao.thirft.domainstruct  ThriftStruct{      1:i32 id;      2:string name;      3:double number;      4:bool isBoolean;}
 

 

 

namespace java com.gqshao.thirft.serviceinclude "ThriftStruct.thrift"service ThriftService {     string helloString(1:string param)       i32 helloInt(1:i32 param)       bool helloBoolean(1:bool param)       void helloVoid()       string helloNull()       i32 add(1:i32 n1, 2:i32 n2)    i32 size(1:list
structs) bool sendSet(1:set
structs) bool sendMap(1:map
structs)}

 

3.生成文件

执行下面的命令生成Domain和Service,并将文件拷贝到源码相应目录下

 

thrift-0.9.2.exe -r -gen java ./ThriftService.thrift
 

 

4.实现接口

 

public class ThriftServiceImpl implements ThriftService.Iface {    private static final Logger logger = LoggerFactory.getLogger(ThriftServiceImpl.class);    @Override    public String helloString(String param) throws TException {        logger.info("helloString : " + param);        return param;    }    @Override    public int helloInt(int param) throws TException {        logger.info("helloInt : " + param + ", begin sleep 20000");        try {            Thread.sleep(20000);        } catch (InterruptedException e) {            e.printStackTrace();        }        logger.info("helloInt : " + param + ", begin sleep over");        return param;    }    @Override    public boolean helloBoolean(boolean param) throws TException {        logger.info("helloBoolean : " + param);        return param;    }    @Override    public void helloVoid() throws TException {        logger.info("helloVoid");    }    @Override    public String helloNull() throws TException {        logger.info("helloNull");        return null;    }    @Override    public int add(int n1, int n2) throws TException {        logger.info("add n1 : " + n1 + ", n2" + n2);        return n1 + n2;    }    @Override    public int size(List
structs) throws TException { logger.info("size"); if (CollectionUtils.isEmpty(structs)) { return 0; } return structs.size(); } @Override public boolean sendSet(Set
structs) throws TException { logger.info("sendSet"); if (CollectionUtils.isEmpty(structs)) { return false; } return true; } @Override public boolean sendMap(Map
structs) throws TException { logger.info("sendMap"); if (MapUtils.isEmpty(structs)) { return false; } return true; }}
 

五.协议

Thrift可以让你选择客户端与服务端之间传输通信协议的类别,在传输协议上总体上划分为文本(text)和二进制(binary)传输协议, 为节约带宽,提供传输效率,一般情况下使用二进制类型的传输协议为多数,但有时会还是会使用基于文本类型的协议,这需要根据项目/产品中的实际需求:
    * TBinaryProtocol – 二进制编码格式进行数据传输。
    * TCompactProtocol – 这种协议非常有效的,使用Variable-Length Quantity (VLQ) 编码对数据进行压缩。
    * TJSONProtocol – 使用JSON的数据编码协议进行数据传输。
    * TSimpleJSONProtocol – 这种节约只提供JSON只写的协议,适用于通过脚本语言解析
    * TDebugProtocol – 在开发的过程中帮助开发人员调试用的,以文本的形式展现方便阅读。

 

六.传输层

常用的传输层有以下几种:

1.TSocket

使用阻塞式 I/O 进行传输,是最常见的模式

 

 

2.TFramedTransport

 

使用非阻塞方式,按块的大小进行传输,类似于 Java 中的 NIO 若使用 TFramedTransport 传输层,其服务器必须修改为非阻塞的服务类型,TNonblockingServerTransport 类是构建非阻塞 socket 的抽象类,TNonblockingServerSocket 类继承 TNonblockingServerTransport  

    使用 TFramedTransport 传输层构建的 HelloServiceServer.java

 

TNonblockingServerTransport serverTransport;  serverTransport = new TNonblockingServerSocket(10005);  Hello.Processor processor = new Hello.Processor(new HelloServiceImpl());  TServer server = new TNonblockingServer(processor, serverTransport);  System.out.println("Start server on port 10005 ...");  server.serve();
 

 

    使用 TFramedTransport 传输层的 HelloServiceClient.java

 

TTransport transport = new TFramedTransport(new TSocket("localhost", 10005));
 

3.TNonblockingTransport

使用非阻塞方式,用于构建异步客户端

 

七.服务端类型

· TSimpleServer 

    TSimplerServer接受一个连接,处理连接请求,直到客户端关闭了连接,它才回去接受一个新的连接。正因为它只在一个单独的线程中以阻塞I/O的方式完成这些工作,所以它只能服务一个客户端连接,其他所有客户端在被服务器端接受之前都只能等待。TSimpleServer主要用于测试目的,不要在生产环境中使用它!

 

· TNonblockingServer

    TNonblockingServer使用非阻塞的I/O解决了TSimpleServer一个客户端阻塞其他所有客户端的问题。它使用了,通过调用,它使得你阻塞在多个连接上,而不是阻塞在单一的连接上。当一或多个连接准备好被接受/读/写时,select()调用便会返回。TNonblockingServer处理这些连接的时候,要么接受它,要么从它那读数据,要么把数据写到它那里,然后再次调用select()来等待下一个可用的连接。通用这种方式,server可同时服务多个客户端,而不会出现一个客户端把其他客户端全部“饿死”的情况。

 

· THsHaServer

    所有消息是被调用select()方法的同一个线程处理的。假设有10个客户端,处理每条消息所需时间为100毫秒,那么,latency和吞吐量分别是多少?当一条消息被处理的时候,其他9个客户端就等着被select,所以客户端需要等待1秒钟才能从服务器端得到回应,吞吐量就是10个请求/秒。如果可以同时处理多条消息的话,会很不错吧?因此,THsHaServer(半同步/半异步的server)就应运而生了。它使用一个单独的线程来处理网络I/O,一个独立的worker线程池来处理消息。这样,只要有空闲的worker线程,消息就会被立即处理,因此多条消息能被并行处理。用上面的例子来说,现在的latency就是100毫秒,而吞吐量就是100个请求/秒。

 

· TThreadedSelectorServer

    Thrift 0.8引入了另一种server实现,即TThreadedSelectorServer。它与THsHaServer的主要区别在于,TThreadedSelectorServer允许你用多个线程来处理网络I/O。它维护了两个线程池,一个用来处理网络I/O,另一个用来进行请求的处理。当网络I/O是瓶颈的时候,TThreadedSelectorServer比THsHaServer的表现要好。

 

· TThreadPoolServer

    TThreadPoolServer与其他三种server不同的是有一个专用的线程用来接受连接。一旦接受了一个连接,它就会被放入ThreadPoolExecutor中的一个worker线程里处理。  worker线程被绑定到特定的客户端连接上,直到它关闭。一旦连接关闭,该worker线程就又回到了线程池中。 你可以配置线程池的最小、最大线程数,默认值分别是5(最小)和Integer.MAX_VALUE(最大)。这意味着,如果有1万个并发的客户端连接,你就需要运行1万个线程。所以它对系统资源的消耗不像其他类型的server一样那么“友好”。此外,如果客户端数量超过了线程池中的最大线程数,在有一个worker线程可用之前,请求将被一直阻塞在那里。TThreadPoolServer的表现非常优异。如果你提前知道了将要连接到你服务器上的客户端数量,并且你不介意运行大量线程的话,TThreadPoolServer对你可能是个很好的选择。

 

结论

TThreadedSelectorServer对大多数案例来说都是个安全之选。如果你的系统资源允许运行大量并发线程的话,你可能会想考虑使用TThreadPoolServer.

 

八.Server和Client

项目中需要引入thrift依赖,Maven依赖如下

org.apache.thrift
libthrift
0.9.2

 

1.服务器端,分别实现了TBinaryProtocol、TCompactProtocol和采用TCompactProtocol协议的ThreadPoolServer 

import com.gqshao.thrift.service.ThriftService;import com.gqshao.thrift.service.impl.ThriftServiceImpl;import org.apache.thrift.TProcessor;import org.apache.thrift.protocol.TBinaryProtocol;import org.apache.thrift.protocol.TCompactProtocol;import org.apache.thrift.server.TServer;import org.apache.thrift.server.TSimpleServer;import org.apache.thrift.server.TThreadPoolServer;import org.apache.thrift.transport.TServerSocket;import org.apache.thrift.transport.TTransportFactory;import org.slf4j.Logger;import org.slf4j.LoggerFactory;public class ThriftServer {    private static final Logger logger = LoggerFactory.getLogger(ThriftServer.class);    /**     * 二进制编码格式进行数据传输     */    public void initTBinaryProtocolServer() {        try {            // transport定义了消息是怎样在客户端和服务器端之间通信的            TServerSocket serverTransport = new TServerSocket(7911);            TSimpleServer.Args args = new TSimpleServer.Args(serverTransport);            // 关联处理器与服务的实现            TProcessor processor = new ThriftService.Processor
(new ThriftServiceImpl()); args.processor(processor); // 使用二进制来编码应用层的数据 protocol(协议)定义了消息是怎样序列化的 args.protocolFactory(new TBinaryProtocol.Factory(true, true)); // 使用普通的socket来传输数据 args.transportFactory(new TTransportFactory()); TServer server = new TSimpleServer(args); logger.info("Start TBinary Protocol Server on port 7911..."); server.serve(); } catch (Exception e) { e.printStackTrace(); } } /** * 高效率的、密集的二进制编码格式进行数据传输 */ public void initTCompactProtocolServer() { try { // transport定义了消息是怎样在客户端和服务器端之间通信的 TServerSocket serverTransport = new TServerSocket(7912); TSimpleServer.Args args = new TSimpleServer.Args(serverTransport); // 关联处理器与服务的实现 TProcessor processor = new ThriftService.Processor
(new ThriftServiceImpl()); args.processor(processor); // 使用二进制来编码应用层的数据 protocol(协议)定义了消息是怎样序列化的 args.protocolFactory(new TCompactProtocol.Factory()); // 使用普通的socket来传输数据 args.transportFactory(new TTransportFactory()); TServer server = new TSimpleServer(args); logger.info("Start TCompact Protocol Server on port 7911..."); server.serve(); } catch (Exception e) { e.printStackTrace(); } } /** * ThreadPoolServer 和 TCompactProtocol */ public void initTThreadPoolServerAndTCompactProtocolServer() { try { TServerSocket serverTransport = new TServerSocket(7913); TThreadPoolServer.Args args = new TThreadPoolServer.Args(serverTransport); // 1.关联处理器与服务的实现 TProcessor processor = new ThriftService.Processor
(new ThriftServiceImpl()); args.processor(processor); // 2.定义protocol 使用高效率的、密集的二进制编码格式进行数据传输 protocol(协议)定义了消息是怎样序列化的 args.protocolFactory(new TCompactProtocol.Factory()); // 使用普通的socket来传输数据 transport定义了消息是怎样在客户端和服务器端之间通信的 // 默认 TTransportFactory // args.transportFactory(new TTransportFactory()); // 实例化TThreadPoolServer TServer server = new TThreadPoolServer(args); logger.info("Start TCompact Protocol Server on port 7911..."); server.serve(); } catch (Exception e) { e.printStackTrace(); } }}
 
import com.google.common.collect.Lists;import com.gqshao.thrift.domain.ThriftStruct;import com.gqshao.thrift.server.ThriftServer;import com.gqshao.thrift.service.ThriftService;import org.apache.thrift.TException;import org.apache.thrift.protocol.TBinaryProtocol;import org.apache.thrift.protocol.TCompactProtocol;import org.apache.thrift.protocol.TProtocol;import org.apache.thrift.transport.TSocket;import org.apache.thrift.transport.TTransport;import org.junit.Test;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import java.util.List;import java.util.concurrent.CountDownLatch;public class ThriftClientService {    private static final Logger logger = LoggerFactory.getLogger(ThriftClientService.class);    public static void initTBinaryProtocolServer() {        new Thread(new Runnable() {            @Override            public void run() {                new ThriftServer().initTBinaryProtocolServer();            }        }).start();    }    @Test    public void testTBinaryProtocolClient() {        TTransport transport = null;        try {            initTBinaryProtocolServer();            transport = new TSocket("localhost", 7911);            transport.open();            // 设置传输协议为 TBinaryProtocol            TProtocol protocol = new TBinaryProtocol(transport);            ThriftService.Client client = new ThriftService.Client(protocol);            // 调用服务的 helloVoid 方法            client.helloString("world");        } catch (TException e) {            e.printStackTrace();        } finally {            if (transport != null) {                transport.close();            }        }    }    public static void initTCompactProtocolServer() {        new Thread(new Runnable() {            @Override            public void run() {                new ThriftServer().initTCompactProtocolServer();            }        }).start();    }    @Test    public void testTCompactProtocolClient() {        TTransport transport = null;        try {            initTCompactProtocolServer();            transport = new TSocket("localhost", 7912);            transport.open();            // 设置传输协议为 TCompactProtocol            TProtocol protocol = new TCompactProtocol(transport);            ThriftService.Client client = new ThriftService.Client(protocol);            // 调用服务的 size 方法            ThriftStruct s1 = new ThriftStruct();            s1.setId(1);            ThriftStruct s2 = new ThriftStruct();            s2.setId(2);            List
structs = Lists.newArrayList(s1, s2); int size = client.size(structs); logger.info("size === " + size); } catch (TException e) { e.printStackTrace(); } finally { if (transport != null) { transport.close(); } } } public static void initTThreadPoolServerAndTCompactProtocolServer() { new Thread(new Runnable() { @Override public void run() { new ThriftServer().initTThreadPoolServerAndTCompactProtocolServer(); } }).start(); } @Test public void testTThreadPoolServerAndTCompactProtocolServerClient() { initTThreadPoolServerAndTCompactProtocolServer(); int threadNum = 50; final CountDownLatch threadSignal = new CountDownLatch(threadNum); try { } catch (Exception e) { e.printStackTrace(); } for (int i = 0; i < 50; i++) { final int finalI = i; new Thread(new Runnable() { @Override public void run() { TTransport transport = null; try { // 传输层选用TSocket transport = new TSocket("localhost", 7913); transport.open(); // 设置传输协议为 TCompactProtocol TProtocol protocol = new TCompactProtocol(transport); ThriftService.Client client = new ThriftService.Client(protocol); // 调用服务的 helloVoid 方法// client.helloString("World !"); client.helloInt(finalI); } catch (Exception e) { e.printStackTrace(); } finally { transport.close(); } threadSignal.countDown(); } }).start(); } try { threadSignal.await(); } catch (InterruptedException e) { e.printStackTrace(); } logger.info("testTThreadPoolServerAndTCompactProtocolServerClient test over"); }}

 

 

 

转载地址:http://hqyci.baihongyu.com/

你可能感兴趣的文章
关于系统崩溃后的Oracle恢复
查看>>
遭遇蚊子的獠牙,想起一种叫黄鸡婆的小虫子
查看>>
使用批处理命令给客户更新oracle数据库
查看>>
回首考研路:那年,那月,还有那条孤独的考研狗
查看>>
GCD小结
查看>>
IOS 第三方库介绍
查看>>
iPhone架构xmpp聊天工具 -xmpp协议初识《一》
查看>>
iOS提交后申请加急审核
查看>>
iOS7单元测试
查看>>
ios framework 通用库的制作
查看>>
出现( linker command failed with exit code 1)错误总结
查看>>
iOS开发中一些常见的并行处理
查看>>
iOS获取手机的Mac地址
查看>>
ios7.1发布企业证书测试包的问题
查看>>
如何自定义iOS中的控件
查看>>
iOS 开发百问
查看>>
Mac环境下svn的使用
查看>>
github简单使用教程
查看>>
如何高效利用GitHub
查看>>
GitHub详细教程
查看>>