官网：

 

简介

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方式的通信：php语言client可以构造一个对象，调用相应的服务方法来调用java语言的服务 ，跨越语言的C/S RPC调用 。

 

Thrift允许定义一个简单的 中的数据类型和服务接口，以作为输入文件， 生成代码用来方便地生成RPC客户端和服务器通信的无缝跨编程语言。

 

安装

使用Homebrew来安装thrift。

－ruby -e "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/master/install)"

－brew install thrift

 

 

一个小实例

 

以UserService为例，描述一下使用thrift的方式，以及其原理。

service.thrift

1. struct User {  
2.     1:i64 id,  
3.     2:string name,  
4.     3:i64 timestamp,  
5.     4:bool vip    
6. }  
7.   
8. service UserService {  
9.     User getById(1:i64 id)  
10. }  

 

你可以将自己的Java服务通过".thrift"文件描述出来，并提供给服务消费端，那么消费端即可以生成自己的API文件。Thrift框架目前已经支持大部分主流的语言。需要注意，因为Thrift考虑到struct/service定义需要兼容多种语言的”风格"，所以它只支持一些基本的数据类型(比如i32,i64,string等)，以及service定义的方法不能重名，即使参数列表不同(并不是所有的语言都能像JAVA一样支持重载)。

 

生成API文件

 

首先下载和安装thrift客户端，比如在windows平台下，下载thrift.exe。不过此处需要提醒，不同的thrift客户端版本生成的API可能不兼容。本例使用thrift-0.9.0.exe，通过"--gen"指定生成API所适配的语言。本实例为生成java客户端API。

Java代码  

1. //windows平台下,将API文件输出在service目录下(此目录需要存在)  
2. > thrift.exe --gen java -o service service.thrift  

 

需要明确的是：Thrift和其他RPC框架不同，thrift在生成的API文件中，已经描述了"调用过程"(即硬编码)，而不是像其他RPC那样在运行时(runtime)动态解析方法调用或者参数。

 

UserService实现类

 

Java代码  

1. public class UserServiceImpl implements UserService.Iface {  
2.     @Override  
3.     public User getById(long id){  
4.         System.out.println("invoke...id:" + id);  
5.         return new User();//for test  
6.     }  
7. }  

 

实现类，需要放在Thrift server端。

 

原理简析

 

User.java

: thrift生成API的能力还是非常的有限，比如在struct中只能使用简单的数据类型(不支持Date,Collection<?>等)，不过我们能从User中看出，它生成的类实现了"Serializable"接口和"TBase”接口。其中Serializable接口表明这个类的实例是需要序列化之后在网络中传输的，为了不干扰Java本身的序列化和反序列化机制，它还重写了readObject和writeObject方法，不过这对thrift本身并没有帮助。

 

TBase接口是thrift序列化和反序列化时使用的，它的两个核心方法：read和write。在上述的thrift文件中，struct定义的每个属性都有一个序号，比如1:id，那么thrift在序列化时，将会根据序号的顺序依次将属性的"名称 + 值"写入inputStream中,反序列化也是如此。(具体参见read和write的实现)。

Java代码  

1. //read方法逐个读取字段,按照"索引",最终将"struct"对象封装完毕.  
2. //write方法也非常类似,按照"索引"顺序逐个输出到流中.  
3. while (true){  
4.         schemeField = iprot.readFieldBegin();  
5.         if (schemeField.type == org.apache.thrift.protocol.TType.STOP) {   
6.           break;  
7.         }  
8.         switch () {  
9.           case 1: // ID  
10.             if (schemeField.type == org.apache.thrift.protocol.TType.I32) {  
11.                = iprot.readI32();  
12.               struct.setIdIsSet(true);  
13.             } else {   
14.               org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);  
15.             }  
16.             break;  
17.           case 2: // NAME  
18.           ..  
19.         }  
20. }  

 

因为thrift的序列化和反序列化实例数据时,是根据"属性序号"进行,这可以保证数据在inputstream和outputstream中顺序是严格的,

此外每个struct中"序号"不能重复,但是可以不需要从"1"开始.如果"序号"有重复,将导致无法生成API文件.这一点也要求API开发者,如果更改了thrift文件中的struct定义,需要重新生成客户端API,否则服务将无法继续使用(可能报错,也可能数据错误).thrift序列化/反序列化的过程和JAVA自带的序列化机制不同,它将不会携带额外的class结构,此外thrift这种序列化机制更加适合网络传输,而且性能更加高效.

 

UserService.Client:  在生成的UserService中，有个Client静态类，这个类就是一个典型的代理类，此类已经实现了UserService的所有方法。开发者需要使用Client类中的API方法与Thrift server端交互，它将负责与Thrift server的Socket链接中，发送请求和接收响应。

 

需要注意的时，每次Client方法调用，都会在一个Socket链接中进行。这就意味着，在使用Client消费服务之前，需要和Thrift server建立有效的TCP链接。(稍后代码示例)

 

1) 发送请求:

Java代码  

1. //参见:TServiceClient  
2. //API方法调用时,发送请求数据流  
3. protected void sendBase(String methodName, TBase args) throws TException {  
4.     oprot_.writeMessageBegin(new TMessage(methodName, TMessageType.CALL, ++seqid_));//首先写入"方法名称"和"seqid_"  
5.     args.write(oprot_);//序列化参数  
6.     oprot_.writeMessageEnd();  
7.     oprot_.getTransport().flush();  
8. }  
9.   
10. protected void receiveBase(TBase result, String methodName) throws TException {  
11.     TMessage msg = iprot_.readMessageBegin();//如果执行有异常  
12.     if (msg.type == TMessageType.EXCEPTION) {  
13.       TApplicationException x = TApplicationException.read(iprot_);  
14.       iprot_.readMessageEnd();  
15.       throw x;  
16.     }//检测seqid是否一致  
17.     if (msg.seqid != seqid_) {  
18.       throw new TApplicationException(TApplicationException.BAD_SEQUENCE_ID, methodName + " failed: out of sequence response");  
19.     }  
20.     result.read(iprot_);//反序列化  
21.     iprot_.readMessageEnd();  
22. }  

 

Thrift提供了简单的容错方式：每次方法调用，都会在Client端标记一个seqid，这是一个自增的本地ID，在TCP请求时将此seqid追加到流中，同时Server端响应时，也将此seqid原样返回过来；这样客户端就可以根据此值用来判断"请求--响应"是对应的，如果出现乱序，将会导致此请求以异常的方式结束。

 

2) 响应

Java代码  

1. //参考: TBaseProcessor.java  
2. @Override  
3. public boolean process(TProtocol in, TProtocol out) throws TException {  
4.     TMessage msg = in.readMessageBegin();  
5.     ProcessFunction fn = processMap.get(msg.name);//根据方法名,查找"内部类"  
6.     if (fn == null) {  
7.       TProtocolUtil.skip(in, TType.STRUCT);  
8.       in.readMessageEnd();  
9.       TApplicationException x = new TApplicationException(TApplicationException.UNKNOWN_METHOD, "Invalid method name: '"+msg.name+"'");  
10.       out.writeMessageBegin(new TMessage(msg.name, TMessageType.EXCEPTION, msg.seqid));  
11.       x.write(out);//序列化响应结果,直接输出  
12.       out.writeMessageEnd();  
13.       out.getTransport().flush();  
14.       return true;  
15.     }  
16.     fn.process(msg.seqid, in, out, iface);  
17.     return true;  
18. }  

 

thrift生成的UserService.Processor类,就是server端用来处理请求过程的"代理类";server端从socket中读取请求需要调用的"方法名" +参数列表,并交付给Processor类处理;和其他的RPC调用不同的时,thrift并没有使用类似于"反射机制"的方式来调用方法,而是将UserService的每个方法生成一个"内部类":

Java代码  

1. public static class getById<I extends Iface> extends org.apache.thrift.ProcessFunction<I, getById_args> {  
2.   public getById() {  
3.     super("getById");//其中getById为标识符  
4.   }  
5.   
6.   public getById_args getEmptyArgsInstance() {  
7.     return new getById_args();  
8.   }  
9.   
10.   protected boolean isOneway() {  
11.     return false;  
12.   }  
13.   //实际处理方法  
14.   public getById_result getResult(I iface, getById_args args) throws org.apache.thrift.TException {  
15.     getById_result result = new getById_result();  
16.     result.success = iface.getById(args.id);  
17.     return result;  
18.   }  
19. }  

 

这个"内部类",将会在Processor初始化的时候,放入到一个map中,此后即可以通过"方法名"查找,然后调用其"getResult"方法了.

Java代码  

1. public static class Processor<I extends Iface> extends org.apache.thrift.TBaseProcessor<I> implements org.apache.thrift.TProcessor {  
2.   
3.     public Processor(I iface) {  
4.       super(iface, getProcessMap(new HashMap<String, org.apache.thrift.ProcessFunction<I, ? extends org.apache.thrift.TBase>>()));  
5.     }  
6.   
7.     protected Processor(I iface, Map<String,  org.apache.thrift.ProcessFunction<I, ? extends  org.apache.thrift.TBase>> processMap) {  
8.       super(iface, getProcessMap(processMap));  
9.     }  
10.   
11.     private static <I extends Iface> Map<String,  org.apache.thrift.ProcessFunction<I, ? extends  org.apache.thrift.TBase>> getProcessMap(Map<String,  org.apache.thrift.ProcessFunction<I, ? extends  org.apache.thrift.TBase>> processMap) {  
12.       //放入map  
13.       processMap.put("getById", new getById());  
14.       return processMap;  
15.     }  
16.     ....  
17. }  

 

3) Server端Socket管理和执行策略

Java代码  

1. TThreadPoolServer  
2. public void serve() {  
3.     try {  
4.       //启动服务  
5.       serverTransport_.listen();  
6.     } catch (TTransportException ttx) {  
7.       LOGGER.error("Error occurred during listening.", ttx);  
8.       return;  
9.     }  
10.   
11.     // Run the preServe event  
12.     if (eventHandler_ != null) {  
13.       eventHandler_.preServe();  
14.     }  
15.   
16.     stopped_ = false;  
17.     setServing(true);  
18.     //循环,直到被关闭  
19.     while (!stopped_) {  
20.       int failureCount = 0;  
21.       try {  
22.         //accept客户端Socket链接,  
23.         //对于每个新链接,将会封装成runnable,并提交给线程或者线程池中运行.  
24.         TTransport client = serverTransport_.accept();  
25.         WorkerProcess wp = new WorkerProcess(client);  
26.         executorService_.execute(wp);  
27.       } catch (TTransportException ttx) {  
28.         if (!stopped_) {  
29.           ++failureCount;  
30.           LOGGER.warn("Transport error occurred during acceptance of message.", ttx);  
31.         }  
32.       }  
33.     }  
34.     //....  
35. }  

 

Thrift Server端，设计思路也非常的直接。当前Service server启动之后，将会以阻塞的方式侦听Socket链接(代码参考TThreadPoolServer)，每建立一个Socket链接，都会将此Socket经过封装之后，放入线程池中，本质上也是一个Socket链接对应一个Worker Thread。这个Thread只会处理此Socket中的所有数据请求，直到Socket关闭。

Java代码  

1. //参考:WorkerProcess  
2. while (true) {  
3.   
4.     if (eventHandler != null) {  
5.       eventHandler.processContext(connectionContext, inputTransport, outputTransport);  
6.     }  
7.   
8.     if(stopped_ || !processor.process(inputProtocol, outputProtocol)) {  
9.       break;  
10.     }  
11. }  

 

当有Socket链接不是很多的时候，TThreadPoolServer并不会有太大的性能问题，可以通过指定ThreadPool中线程的个数进行简单的调优。如果Socket链接很多，我们只能使用TThreadedSelectorServer来做支撑，TThreadedSelectorServer内部基于NIO模式，具有异步的特性，可以极大的提升server端的并发能力；不过在绝大多数情况下，在thrift中使用"异步"似乎不太容易让人接受，毕竟这意味着Client端需要阻塞，并且在高并发环境中这个阻塞时间是不可控的。但SelecorServer确实可以有效的提升Server的并发能力，而且在一定程度上可以提升吞吐能力，这或许是我们优化Thrift Server比较可靠的方式之一。

 

Client端代码示例

Java代码  

1. public class UserServiceClient {  
2.   
3.     public void startClient() {  
4.         TTransport transport;  
5.         try {  
6.             transport = new TSocket("localhost", 1234);  
7.             TProtocol protocol = new TBinaryProtocol(transport);  
8.             UserService.Client client = new UserService.Client(protocol);  
9.             transport.open();  
10.             User user = client.getById(1000);  
11.               
12.             transport.close();  
13.         } catch (TTransportException e) {  
14.             e.printStackTrace();  
15.         } catch (TException e) {  
16.             e.printStackTrace();  
17.         }  
18.     }  
19.   
20. }  

 

Server端代码示例 

Java代码  

1. public class Server {  
2.     public void startServer() {  
3.         try {  
4.             TServerSocket serverTransport = new TServerSocket(1234);  
5.             UserService.Processor process = new Processor(new UserServiceImpl());  
6.             Factory portFactory = new TBinaryProtocol.Factory(true, true);  
7.             Args args = new Args(serverTransport);  
8.             args.processor(process);  
9.             args.protocolFactory(portFactory);  
10.             TServer server = new TThreadPoolServer(args);  
11.             server.serve();  
12.         } catch (TTransportException e) {  
13.             e.printStackTrace();  
14.         }  
15.     }  
16. }  

 

到这里，你就会发现，一个service，需要server端启动一个ServerSocket，如果你有很多service，那么你需要让这些service尽可能的分布在不同的物理server上，否则一个物理server上运行太多的ServerSocket进程并不是一件让人愉快的事情.。或者你让几个service整合成一个。

 

问题总没有想象的那么简单，其实service被拆分的粒度越细，越容易被部署和扩展，对于负载均衡就更加有利。如何让一个service分布式部署，稍后再继续分享。

 

总结

1) Thrift文件定义struct和service API，此文件可以被其他语言生成API文件或者类文件。

2) 使用Thrift客户端生成API文件。

3) Java服务端(即服务提供端)，实现service功能。

4) 服务端将server发布成一个Thrift server：即将service嵌入到一个ServerSocket中。

5) 客户端启动Socket，并和Thrift server建立TCP连接，并使用Client代理类操作远程接口。

 

服务端开发

 

Thrift服务server端，其实就是一个ServerSocket线程+处理器，当Thrift-client端建立链接之后，处理器负责解析socket流信息，并根据其指定的"方法名"+参数列表，来调用"服务实现类”的方法，并将执行结果(或者异常)写入到socket中。

 

一个server，就需要创建一个ServerSocket，并侦听本地的一个端口，这种情况对分布式部署，有一些额外的要求：client端需要知道一个"service"被部署在了那些server上。

 

设计思路:

1) 每个server内部采用threadPool的方式,来提升并发能力.

2) 当server启动成功后,向zookeeper注册服务节点,此后client端就可以"感知到"服务的状态

3) 通过spring的方式,配置thrift-server服务类.

其中x注册是可选选项

 

1.pom.xml

 

1. <dependencies>  
2.     <dependency>  
3.         <groupId>org.springframework</groupId>  
4.         <artifactId>spring-context</artifactId>  
5.         <version>3.0.7.RELEASE</version>  
6.     </dependency>  
7.     <dependency>  
8.         <groupId>org.apache.zookeeper</groupId>  
9.         <artifactId>zookeeper</artifactId>  
10.         <version>3.4.5</version>  
11.         <!--<exclusions>-->  
12.             <!--<exclusion>-->  
13.                 <!--<groupId>log4j</groupId>-->  
14.                 <!--<artifactId>log4j</artifactId>-->  
15.             <!--</exclusion>-->  
16.         <!--</exclusions>-->  
17.     </dependency>  
18.     <!--  
19.     <dependency>  
20.         <groupId>com.101tec</groupId>  
21.         <artifactId>zkclient</artifactId>  
22.         <version>0.4</version>  
23.     </dependency>  
24.     -->  
25.     <dependency>  
26.         <groupId>org.apache.thrift</groupId>  
27.         <artifactId>libthrift</artifactId>  
28.         <version>0.9.1</version>  
29.     </dependency>  
30.     <dependency>  
31.         <groupId>org.apache.curator</groupId>  
32.         <artifactId>curator-recipes</artifactId>  
33.         <version>2.3.0</version>  
34.     </dependency>  
35.     <dependency>  
36.         <groupId>commons-pool</groupId>  
37.         <artifactId>commons-pool</artifactId>  
38.         <version>1.6</version>  
39.     </dependency>  
40.   
41. </dependencies>  

 

本实例,使用了apache-curator作为zookeeper客户端.

 

2. spring-thrift-server.xml

 

1. <!-- zookeeper -->  
2. <bean id="thriftZookeeper" class="com.demo.thrift.zookeeper.ZookeeperFactory" destroy-method="close">  
3.     <property name="connectString" value="127.0.0.1:2181"></property>  
4.     <property name="namespace" value="demo/thrift-service"></property>  
5. </bean>  
6. <bean id="sericeAddressReporter" class="com.demo.thrift.support.impl.DynamicAddressReporter" destroy-method="close">  
7.     <property name="zookeeper" ref="thriftZookeeper"></property>  
8. </bean>  
9. <bean id="userService" class="com.demo.service.UserServiceImpl"/>  
10. <bean class="com.demo.thrift.ThriftServiceServerFactory" destroy-method="close">  
11.     <property name="service" ref="userService"></property>  
12.     <property name="configPath" value="UserServiceImpl"></property>  
13.     <property name="port" value="9090"></property>  
14.     <property name="addressReporter" ref="sericeAddressReporter"></property>  
15. </bean>  

 

3. ThriftServiceServerFactory.java

 

此类严格上说并不是一个工厂类,它的主要作用就是封装指定的"service" ,然后启动一个server的过程,其中"service"属性表示服务的实现类,addressReporter表示当server启动成功后,需要指定的操作(比如,向zookeeper发送service的IP信息).

 

究竟当前server的ip地址是多少,在不同的设计中,有所不同,比如:有些管理员喜欢将本机的IP地址写入到os下的某个文件中,如果上层应用需要获取可靠的IP信息,就需要读取这个文件...你可以实现自己的ThriftServerIpTransfer来获取当前server的IP.

 

为了减少xml中的配置信息,在factory中,使用了反射机制来构建"Processor"类.

 

1. public class ThriftServiceServerFactory implements InitializingBean {  
2.   
3.     private Integer port;  
4.   
5.     private Integer priority = 1;// default  
6.   
7.     private Object service;// serice实现类  
8.   
9.     private ThriftServerIpTransfer ipTransfer;  
10.   
11.     private ThriftServerAddressReporter addressReporter;  
12.       
13.     private ServerThread serverThread;  
14.       
15.     private String configPath;  
16.   
17.     public void setService(Object service) {  
18.         this.service = service;  
19.     }  
20.   
21.     public void setPriority(Integer priority) {  
22.         this.priority = priority;  
23.     }  
24.   
25.     public void setPort(Integer port) {  
26.         this.port = port;  
27.     }  
28.   
29.     public void setIpTransfer(ThriftServerIpTransfer ipTransfer) {  
30.         this.ipTransfer = ipTransfer;  
31.     }  
32.   
33.     public void setAddressReporter(ThriftServerAddressReporter addressReporter) {  
34.         this.addressReporter = addressReporter;  
35.     }  
36.       
37.   
38.     public void setConfigPath(String configPath) {  
39.         this.configPath = configPath;  
40.     }  
41.   
42.     @Override  
43.     public void afterPropertiesSet() throws Exception {  
44.         if (ipTransfer == null) {  
45.             ipTransfer = new LocalNetworkIpTransfer();  
46.         }  
47.         String ip = ipTransfer.getIp();  
48.         if (ip == null) {  
49.             throw new NullPointerException("cant find server ip...");  
50.         }  
51.         String hostname = ip + ":" + port + ":" + priority;  
52.         Class serviceClass = service.getClass();  
53.         ClassLoader classLoader = Thread.currentThread().getContextClassLoader();  
54.         Class<?>[] interfaces = serviceClass.getInterfaces();  
55.         if (interfaces.length == 0) {  
56.             throw new IllegalClassFormatException("service-class should implements Iface");  
57.         }  
58.   
59.         // reflect,load "Processor";  
60.         Processor processor = null;  
61.         for (Class clazz : interfaces) {  
62.             String cname = clazz.getSimpleName();  
63.             if (!cname.equals("Iface")) {  
64.                 continue;  
65.             }  
66.             String pname = clazz.getEnclosingClass().getName() + "$Processor";  
67.             try {  
68.                 Class pclass = classLoader.loadClass(pname);  
69.                 if (!pclass.isAssignableFrom(Processor.class)) {  
70.                     continue;  
71.                 }  
72.                 Constructor constructor = pclass.getConstructor(clazz);  
73.                 processor = (Processor) constructor.newInstance(service);  
74.                 break;  
75.             } catch (Exception e) {  
76.                 //  
77.             }  
78.         }  
79.   
80.         if (processor == null) {  
81.             throw new IllegalClassFormatException("service-class should implements Iface");  
82.         }  
83.         //需要单独的线程,因为serve方法是阻塞的.  
84.         serverThread = new ServerThread(processor, port);  
85.         serverThread.start();  
86.         // report  
87.         if (addressReporter != null) {  
88.             addressReporter.report(configPath, hostname);  
89.         }  
90.   
91.     }  
92.   
93.     class ServerThread extends Thread {  
94.         private TServer server;  
95.   
96.         ServerThread(Processor processor, int port) throws Exception {  
97.             TServerSocket serverTransport = new TServerSocket(port);  
98.             Factory portFactory = new TBinaryProtocol.Factory(true, true);  
99.             Args args = new Args(serverTransport);  
100.             args.processor(processor);  
101.             args.protocolFactory(portFactory);  
102.             server = new TThreadPoolServer(args);  
103.         }  
104.   
105.         @Override  
106.         public void run(){  
107.             try{  
108.                 server.serve();  
109.             }catch(Exception e){  
110.                 //  
111.             }  
112.         }  
113.           
114.         public void stopServer(){  
115.             server.stop();  
116.         }  
117.     }  
118.   
119.     public void close() {  
120.         serverThread.stopServer();  
121.     }  
122.   
123. }  

 

4. DynamicAddressReporter.java

 

在ThriftServiceServerFactory中,有个可选的属性:addressReporter, DynamicAddressReporter提供了向zookeeper注册service信息的能力,当server启动正常后,把server的IP + port发送到zookeeper中;那么此后服务消费client,就可以从zookeeper中获取server列表,并与它们建立链接(池).这样client端只需要关注zookeeper的节点名称即可,不需要配置大量的ip+port.

1. public class DynamicAddressReporter implements ThriftServerAddressReporter {  
2.       
3.     private CuratorFramework zookeeper;  
4.       
5.     public DynamicAddressReporter(){}  
6.       
7.     public DynamicAddressReporter(CuratorFramework zookeeper){  
8.         this.zookeeper = zookeeper;  
9.     }  
10.   
11.   
12.     public void setZookeeper(CuratorFramework zookeeper) {  
13.         this.zookeeper = zookeeper;  
14.     }  
15.   
16.     @Override  
17.     public void report(String service, String address) throws Exception {  
18.         if(zookeeper.getState() == CuratorFrameworkState.LATENT){  
19.             zookeeper.start();  
20.             zookeeper.newNamespaceAwareEnsurePath(service);  
21.         }  
22.         zookeeper.create()  
23.             .creatingParentsIfNeeded()  
24.             .withMode(CreateMode.EPHEMERAL_SEQUENTIAL)  
25.             .forPath(service +"/i_",address.getBytes("utf-8"));  
26.     }  
27.       
28.       
29.     public void close(){  
30.         zookeeper.close();  
31.     }  
32.   
33. }  

5. 测试类

1. public class ServiceMain {  
2.   
3.     /** 
4.      * @param args 
5.      */  
6.     public static void main(String[] args) {  
7.         try {  
8.             ApplicationContext context = new ClassPathXmlApplicationContext("spring-thrift-server.xml");  
9.             Thread.sleep(3000000);  
10.         } catch (Exception e) {  
11.             e.printStackTrace();  
12.         }  
13.   
14.     }  
15.   
16. }  

 

客户端开发

 

Thrift-client作为服务消费端，由于thrift使用socket通讯，因此它需要面对几个问题：

1) client端需要知道server端的IP+port，如果是分布式部署，还需要知道所有server的IP+port列表。

2) client为了提升性能，不可能只使用一个socket来处理并发请求，当然也不能每个请求都创建一个socket；我们需要使用连接池方案。

3) 对于java开发工程师而言，基于spring配置thrift服务，可以提供很多的便利。

4) 基于zookeeper配置管理，那么client端就不需要"硬编码"的配置server的ip+port，可以使用zookeeper来推送每个service的服务地址。

5) 因为thrift-client端不使用连接池的话，将不能有效的提高并发能力，本文重点描述看如何使用thrift-client连接池。

 

1. pom.xml

 

1. <dependencies>  
2.     <dependency>  
3.         <groupId>org.springframework</groupId>  
4.         <artifactId>spring-context</artifactId>  
5.         <version>3.0.7.RELEASE</version>  
6.     </dependency>  
7.     <dependency>  
8.         <groupId>org.apache.zookeeper</groupId>  
9.         <artifactId>zookeeper</artifactId>  
10.         <version>3.4.5</version>  
11.         <!--<exclusions>-->  
12.             <!--<exclusion>-->  
13.                 <!--<groupId>log4j</groupId>-->  
14.                 <!--<artifactId>log4j</artifactId>-->  
15.             <!--</exclusion>-->  
16.         <!--</exclusions>-->  
17.     </dependency>  
18.     <!--  
19.     <dependency>  
20.         <groupId>com.101tec</groupId>  
21.         <artifactId>zkclient</artifactId>  
22.         <version>0.4</version>  
23.     </dependency>  
24.     -->  
25.     <dependency>  
26.         <groupId>org.apache.thrift</groupId>  
27.         <artifactId>libthrift</artifactId>  
28.         <version>0.9.1</version>  
29.     </dependency>  
30.     <dependency>  
31.         <groupId>org.apache.curator</groupId>  
32.         <artifactId>curator-recipes</artifactId>  
33.         <version>2.3.0</version>  
34.     </dependency>  
35.     <dependency>  
36.         <groupId>commons-pool</groupId>  
37.         <artifactId>commons-pool</artifactId>  
38.         <version>1.6</version>  
39.     </dependency>  
40.   
41. </dependencies>  

2. spring-thrift-client.xml

其中zookeeper作为可选项,开发者也可以通过制定serverAddress的方式指定server的地址.

 

1. <!-- fixedAddress -->  
2. <!--    
3. <bean id="userService" class="com.demo.thrift.ThriftServiceClientProxyFactory">  
4.     <property name="service" value="com.demo.service.UserService"></property>  
5.     <property name="serverAddress" value="127.0.0.1:9090:2"></property>  
6.     <property name="maxActive" value="5"></property>  
7.     <property name="idleTime" value="10000"></property>  
8. </bean>  
9. -->  
10. <!-- zookeeper -->  
11. <bean id="thriftZookeeper" class="com.demo.thrift.zookeeper.ZookeeperFactory" destroy-method="close">  
12.     <property name="connectString" value="127.0.0.1:2181"></property>  
13.     <property name="namespace" value="demo/thrift-service"></property>  
14. </bean>  
15. <bean id="userService" class="com.demo.thrift.ThriftServiceClientProxyFactory" destroy-method="close">  
16.     <property name="service" value="com.demo.service.UserService"></property>  
17.     <property name="maxActive" value="5"></property>  
18.     <property name="idleTime" value="1800000"></property>  
19.     <property name="addressProvider">  
20.         <bean class="com.demo.thrift.support.impl.DynamicAddressProvider">  
21.             <property name="configPath" value="UserServiceImpl"></property>  
22.             <property name="zookeeper" ref="thriftZookeeper"></property>  
23.         </bean>  
24.     </property>  
25. </bean>  

 

3. ThriftServiceClientProxyFactory.java

因为我们要在client端使用连接池方案,那么就需要对client的方法调用过程,进行代理,这个类,就是维护了一个"Client"代理类,并在方法调用时,从"对象池"中取出一个"Client"对象,并在方法实际调用结束后归还给"对象池".  

1. @SuppressWarnings("rawtypes")  
2. public class ThriftServiceClientProxyFactory implements FactoryBean,InitializingBean {  
3.   
4.     private String service;  
5.   
6.     private String serverAddress;  
7.       
8.     private Integer maxActive = 32;//最大活跃连接数  
9.       
10.     ms,default 3 min,链接空闲时间  
11.     //-1,关闭空闲检测  
12.     private Integer idleTime = 180000;  
13.     private ThriftServerAddressProvider addressProvider;  
14.   
15.     private Object proxyClient;  
16.       
17.   
18.     public void setMaxActive(Integer maxActive) {  
19.         this.maxActive = maxActive;  
20.     }  
21.   
22.   
23.     public void setIdleTime(Integer idleTime) {  
24.         this.idleTime = idleTime;  
25.     }  
26.   
27.   
28.     public void setService(String service) {  
29.         this.service = service;  
30.     }  
31.   
32.   
33.     public void setServerAddress(String serverAddress) {  
34.         this.serverAddress = serverAddress;  
35.     }  
36.   
37.   
38.     public void setAddressProvider(ThriftServerAddressProvider addressProvider) {  
39.         this.addressProvider = addressProvider;  
40.     }  
41.   
42.     private Class objectClass;  
43.       
44.     private GenericObjectPool<TServiceClient> pool;  
45.       
46.     private PoolOperationCallBack callback = new PoolOperationCallBack() {  
47.           
48.         @Override  
49.         public void make(TServiceClient client) {  
50.             System.out.println("create");  
51.               
52.         }  
53.           
54.         @Override  
55.         public void destroy(TServiceClient client) {  
56.             System.out.println("destroy");  
57.               
58.         }  
59.     };  
60.   
61.     @Override  
62.     public void afterPropertiesSet() throws Exception {  
63.         if(serverAddress != null){  
64.             addressProvider = new FixedAddressProvider(serverAddress);  
65.         }  
66.         ClassLoader classLoader = Thread.currentThread().getContextClassLoader();  
67.         //加载Iface接口  
68.         objectClass = classLoader.loadClass(service + "$Iface");  
69.         //加载Client.Factory类  
70.         Class<TServiceClientFactory<TServiceClient>> fi = (Class<TServiceClientFactory<TServiceClient>>)classLoader.loadClass(service + "$Client$Factory");  
71.         TServiceClientFactory<TServiceClient> clientFactory = fi.newInstance();  
72.         ThriftClientPoolFactory clientPool = new ThriftClientPoolFactory(addressProvider, clientFactory,callback);  
73.         GenericObjectPool.Config poolConfig = new GenericObjectPool.Config();  
74.         poolConfig.maxActive = maxActive;  
75.         poolConfig.minIdle = 0;  
76.         poolConfig.minEvictableIdleTimeMillis = idleTime;  
77.         poolConfig.timeBetweenEvictionRunsMillis = idleTime/2L;  
78.         pool = new GenericObjectPool<TServiceClient>(clientPool,poolConfig);  
79.         proxyClient = Proxy.newProxyInstance(classLoader,new Class[]{objectClass},new InvocationHandler() {  
80.             @Override  
81.             public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {  
82.                 //  
83.                 TServiceClient client = pool.borrowObject();  
84.                 try{  
85.                     return method.invoke(client, args);  
86.                 }catch(Exception e){  
87.                     throw e;  
88.                 }finally{  
89.                     pool.returnObject(client);  
90.                 }  
91.             }  
92.         });  
93.     }  
94.   
95.     @Override  
96.     public Object getObject() throws Exception {  
97.         return proxyClient;  
98.     }  
99.   
100.     @Override  
101.     public Class<?> getObjectType() {  
102.         return objectClass;  
103.     }  
104.   
105.     @Override  
106.     public boolean isSingleton() {  
107.         return true;  //To change body of implemented methods use File | Settings | File Templates.  
108.     }  
109.       
110.     public void close(){  
111.         if(addressProvider != null){  
112.             addressProvider.close();  
113.         }  
114.     }  
115. }  

4. ThriftClientPoolFactory.java

"Client"对象池,对象池中是已经实例化的Client对象,Client对象负责与Thrift server通信.

1. /** 
2.  * 连接池,thrift-client for spring 
3.  */  
4. public class ThriftClientPoolFactory extends BasePoolableObjectFactory<TServiceClient>{  
5.   
6.     private final ThriftServerAddressProvider addressProvider;  
7.       
8.     private final TServiceClientFactory<TServiceClient> clientFactory;  
9.       
10.     private PoolOperationCallBack callback;  
11.   
12.     protected ThriftClientPoolFactory(ThriftServerAddressProvider addressProvider,TServiceClientFactory<TServiceClient> clientFactory) throws Exception {  
13.         this.addressProvider = addressProvider;  
14.         this.clientFactory = clientFactory;  
15.     }  
16.       
17.     protected ThriftClientPoolFactory(ThriftServerAddressProvider addressProvider,TServiceClientFactory<TServiceClient> clientFactory,PoolOperationCallBack callback) throws Exception {  
18.         this.addressProvider = addressProvider;  
19.         this.clientFactory = clientFactory;  
20.         this.callback = callback;  
21.     }  
22.   
23.   
24.   
25.     @Override  
26.     public TServiceClient makeObject() throws Exception {  
27.         InetSocketAddress address = addressProvider.selector();  
28.         TSocket tsocket = new TSocket(address.getHostName(),address.getPort());  
29.         TProtocol protocol = new TBinaryProtocol(tsocket);  
30.         TServiceClient client = this.clientFactory.getClient(protocol);  
31.         tsocket.open();  
32.         if(callback != null){  
33.             try{  
34.                 callback.make(client);  
35.             }catch(Exception e){  
36.                 //  
37.             }  
38.         }  
39.         return client;  
40.     }  
41.   
42.     public void destroyObject(TServiceClient client) throws Exception {  
43.         if(callback != null){  
44.             try{  
45.                 callback.destroy(client);  
46.             }catch(Exception e){  
47.                 //  
48.             }  
49.         }  
50.         TTransport pin = client.getInputProtocol().getTransport();  
51.         pin.close();  
52.     }  
53.   
54.     public boolean validateObject(TServiceClient client) {  
55.         TTransport pin = client.getInputProtocol().getTransport();  
56.         return pin.isOpen();  
57.     }  
58.       
59.     static interface PoolOperationCallBack {  
60.         //销毁client之前执行  
61.         void destroy(TServiceClient client);  
62.         //创建成功是执行  
63.         void make(TServiceClient client);  
64.     }  
65.   
66. }  

5. DynamicAddressProvider.java

将zookeeper作为server地址的提供者,这样客户端就不需要再配置文件中指定一堆ip + port,而且当server服务有更新时,也不需要client端重新配置.

1. /** 
2.  * 可以动态获取address地址,方案设计参考 
3.  * 1) 可以间歇性的调用一个web-service来获取地址 
4.  * 2) 可以使用事件监听机制,被动的接收消息,来获取最新的地址(比如基于MQ,nio等) 
5.  * 3) 可以基于zookeeper-watcher机制,获取最新地址 
6.  * <p/> 
7.  * 本实例,使用zookeeper作为"config"中心,使用apache-curator方法库来简化zookeeper开发 
8.  * 如下实现,仅供参考 
9.  */  
10. public class DynamicAddressProvider implements ThriftServerAddressProvider, InitializingBean {  
11.   
12.     private String configPath;  
13.   
14.     private PathChildrenCache cachedPath;  
15.   
16.     private CuratorFramework zookeeper;  
17.       
18.     //用来保存当前provider所接触过的地址记录  
19.     //当zookeeper集群故障时,可以使用trace中地址,作为"备份"  
20.     private Set<String> trace = new HashSet<String>();  
21.   
22.     private final List<InetSocketAddress> container = new ArrayList<InetSocketAddress>();  
23.   
24.     private Queue<InetSocketAddress> inner = new LinkedList<InetSocketAddress>();  
25.       
26.     private Object lock = new Object();  
27.       
28.     private static final Integer DEFAULT_PRIORITY = 1;  
29.   
30.     public void setConfigPath(String configPath) {  
31.         this.configPath = configPath;  
32.     }  
33.   
34.     public void setZookeeper(CuratorFramework zookeeper) {  
35.         this.zookeeper = zookeeper;  
36.     }  
37.   
38.     @Override  
39.     public void afterPropertiesSet() throws Exception {  
40.         //如果zk尚未启动,则启动  
41.         if(zookeeper.getState() == CuratorFrameworkState.LATENT){  
42.             zookeeper.start();  
43.         }  
44.         buildPathChildrenCache(zookeeper, configPath, true);  
45.         cachedPath.start(StartMode.POST_INITIALIZED_EVENT);  
46.     }  
47.   
48.     private void buildPathChildrenCache(CuratorFramework client, String path, Boolean cacheData) throws Exception {  
49.         cachedPath = new PathChildrenCache(client, path, cacheData);  
50.         cachedPath.getListenable().addListener(new PathChildrenCacheListener() {  
51.             @Override  
52.             public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {  
53.                 PathChildrenCacheEvent.Type eventType = event.getType();  
54.                 switch (eventType) {  
55. //                    case CONNECTION_RECONNECTED:  
56. //                          
57. //                        break;  
58.                     case CONNECTION_SUSPENDED:  
59.                     case CONNECTION_LOST:  
60.                         System.out.println("Connection error,waiting...");  
61.                         return;  
62.                     default:  
63.                         //  
64.                 }  
65.                 //任何节点的时机数据变动,都会rebuild,此处为一个"简单的"做法.  
66.                 cachedPath.rebuild();  
67.                 rebuild();  
68.             }  
69.               
70.             protected void rebuild() throws Exception {  
71.                 List<ChildData> children = cachedPath.getCurrentData();  
72.                 if (children == null || children.isEmpty()) {  
73.                     //有可能所有的thrift server都与zookeeper断开了链接  
74.                     //但是,有可能,thrift client与thrift server之间的网络是良好的  
75.                     //因此此处是否需要清空container,是需要多方面考虑的.  
76.                     container.clear();  
77.                     System.out.println("thrift server-cluster error....");  
78.                     return;  
79.                 }  
80.                 List<InetSocketAddress> current = new ArrayList<InetSocketAddress>();  
81.                 for (ChildData data : children) {  
82.                     String address = new String(data.getData(), "utf-8");  
83.                     current.addAll(transfer(address));  
84.                     trace.add(address);  
85.                 }  
86.                 Collections.shuffle(current);  
87.                 synchronized (lock) {  
88.                     container.clear();  
89.                     container.addAll(current);  
90.                     inner.clear();  
91.                     inner.addAll(current);  
92.                       
93.                 }  
94.             }  
95.         });  
96.     }  
97.     
98.     private List<InetSocketAddress> transfer(String address){  
99.         String[] hostname = address.split(":");  
100.         Integer priority = DEFAULT_PRIORITY;  
101.         if (hostname.length == 3) {  
102.             priority = Integer.valueOf(hostname[2]);  
103.         }  
104.         String ip = hostname[0];  
105.         Integer port = Integer.valueOf(hostname[1]);  
106.         List<InetSocketAddress> result = new ArrayList<InetSocketAddress>();  
107.         for (int i = 0; i < priority; i++) {  
108.             result.add(new InetSocketAddress(ip, port));  
109.         }  
110.         return result;  
111.     }  
112.   
113.   
114.     @Override  
115.     public List<InetSocketAddress> getAll() {  
116.         return Collections.unmodifiableList(container);  
117.     }  
118.   
119.     @Override  
120.     public synchronized InetSocketAddress selector() {  
121.         if (inner.isEmpty()) {  
122.             if(!container.isEmpty()){  
123.                 inner.addAll(container);  
124.             }else if(!trace.isEmpty()){  
125.                 synchronized (lock) {  
126.                     for(String hostname : trace){  
127.                         container.addAll(transfer(hostname));  
128.                     }  
129.                     Collections.shuffle(container);  
130.                     inner.addAll(container);  
131.                 }  
132.             }  
133.         }  
134.         return inner.poll();//null  
135.     }  
136.   
137.   
138.     @Override  
139.     public void close() {  
140.         try {  
141.             cachedPath.close();  
142.             zookeeper.close();  
143.         } catch (Exception e) {  
144.             //  
145.         }  
146.     }  
147. }  

 

到此为止，我们的Thrift基本上就可以顺利运行起来了。

 

使用ZooKeeper构建集群

通用办法是使用apache-curator组件来支持thrift连接zk提供集群服务并且推送服务信息变化。

 

其他高级话题

 

对于Thrift服务化的改造，主要是客户端，可以从如下几个方面进行：

 

1.服务端的服务注册，客户端自动发现，无需手工修改配置，这里我们使用zookeeper，但由于zookeeper本身提供的客户端使用较为复杂，因此采用curator-recipes工具类进行处理服务的注册与发现。

2.客户端使用连接池对服务调用进行管理，提升性能，这里我们使用Apache Commons项目commons-pool，可以大大减少代码的复杂度。

3.关于Failover/LoadBalance，由于zookeeper的watcher，当服务端不可用是及时通知客户端，并移除不可用的服务节点，而LoadBalance有很多算法，这里我们采用随机加权方式，也是常有的负载算法，至于其他的算法介绍参考: 。

4.使thrift服务的注册和发现可以基于spring配置,可以提供很多的便利。

5.其他的改造如：

1）通过动态代理实现client和server端的交互细节透明化，让用户只需通过服务方提供的接口进行访问

2）Thrift通过两种方式调用服务Client和Iface

1. // *) Client API 调用  
2. (EchoService.Client)client.echo("hello lilei");  ---(1)  
3. // *) Service 接口 调用  
4. (EchoService.Iface)service.echo("hello lilei");  ---(2)  

 

Client API的方式, 不推荐, 我们推荐Service接口的方式(服务化)。

 

下面我们来一一实现：

一、pom.xml引入依赖jar包

1. <dependency>  
2.             <groupId>org.apache.thrift</groupId>  
3.             <artifactId>libthrift</artifactId>  
4.             <version>0.9.2</version>  
5.         </dependency>  
6.         <dependency>  
7.             <groupId>commons-pool</groupId>  
8.             <artifactId>commons-pool</artifactId>  
9.             <version>1.6</version>  
10.         </dependency>  
11.         <dependency>  
12.             <groupId>org.springframework</groupId>  
13.             <artifactId>spring-context</artifactId>  
14.             <version>4.0.9.RELEASE</version>  
15.         </dependency>  
16.   
17.         <dependency>  
18.             <groupId>org.apache.zookeeper</groupId>  
19.             <artifactId>zookeeper</artifactId>  
20.             <version>3.4.6</version>  
21.         </dependency>  
22.         <dependency>  
23.             <groupId>org.apache.curator</groupId>  
24.             <artifactId>curator-recipes</artifactId>  
25.             <version>2.7.1</version>  
26.         </dependency>  

 

二、使用zookeeper管理服务节点配置

RPC服务往平台化的方向发展, 会屏蔽掉更多的服务细节(服务的IP地址集群, 集群的扩容和迁移), 只暴露服务接口. 这部分的演化, 使得server端和client端完全的解耦合. 两者的交互通过ConfigServer(MetaServer)的中介角色来搭线。

注: 该图源自dubbo的官网

这边借助Zookeeper来扮演该角色, server扮演发布者的角色, 而client扮演订阅者的角色.

Zookeeper是分布式应用协作服务. 它实现了paxos的一致性算法, 在命名管理/配置推送/数据同步/主从切换方面扮演重要的角色。 其数据组织类似文件系统的目录结构: 

每个节点被称为znode, 为znode节点依据其特性, 又可以分为如下类型:

　　1). PERSISTENT: 永久节点

　　2). EPHEMERAL: 临时节点, 会随session(client disconnect)的消失而消失

　　3). PERSISTENT_SEQUENTIAL: 永久节点, 其节点的名字编号是单调递增的

　　4). EPHEMERAL_SEQUENTIAL: 临时节点, 其节点的名字编号是单调递增的

　　注: 临时节点不能成为父节点

　　Watcher观察模式, client可以注册对节点的状态/内容变更的事件回调机制. 其Event事件的两类属性需要关注下:

　　1). KeeperState: Disconnected,SyncConnected,Expired

　　2). EventType: None,NodeCreated,NodeDeleted,NodeDataChanged,NodeChildrenChanged

RPC服务端:

　　作为具体业务服务的RPC服务发布方, 对其自身的服务描述由以下元素构成.

　　1). namespace: 命名空间，来区分不同应用 

　　2). service: 服务接口, 采用发布方的类全名来表示

　　3). version: 版本号

　　借鉴了Maven的GAV坐标系, 三维坐标系更符合服务平台化的大环境. 

　　*) 数据模型的设计

　　具体RPC服务的注册路径为: /rpc/{namespace}/{service}/{version}, 该路径上的节点都是永久节点

　　RPC服务集群节点的注册路径为: /rpc/{namespace}/{service}/{version}/{ip:port:weight}, 末尾的节点是临时节点.

1.定义Zookeeper的客户端的管理

ZookeeperFactory.java

 

package cn.slimsmart.thrift.rpc.zookeeper;  

1.   
2. import org.apache.curator.framework.CuratorFramework;  
3. import org.apache.curator.framework.CuratorFrameworkFactory;  
4. import org.apache.curator.retry.ExponentialBackoffRetry;  
5. import org.springframework.beans.factory.FactoryBean;  
6. import org.springframework.util.StringUtils;  
7.   
8. /** 
9.  * 获取zookeeper客户端链接 
10.  */  
11. public class ZookeeperFactory implements FactoryBean<CuratorFramework> {  
12.   
13.     private String zkHosts;  
14.     // session超时  
15.     private int sessionTimeout = 30000;  
16.     private int connectionTimeout = 30000;  
17.   
18.     // 共享一个zk链接  
19.     private boolean singleton = true;  
20.   
21.     // 全局path前缀,常用来区分不同的应用  
22.     private String namespace;  
23.   
24.     private final static String ROOT = "rpc";  
25.   
26.     private CuratorFramework zkClient;  
27.   
28.     public void setZkHosts(String zkHosts) {  
29.         this.zkHosts = zkHosts;  
30.     }  
31.   
32.     public void setSessionTimeout(int sessionTimeout) {  
33.         this.sessionTimeout = sessionTimeout;  
34.     }  
35.   
36.     public void setConnectionTimeout(int connectionTimeout) {  
37.         this.connectionTimeout = connectionTimeout;  
38.     }  
39.   
40.     public void setSingleton(boolean singleton) {  
41.         this.singleton = singleton;  
42.     }  
43.   
44.     public void setNamespace(String namespace) {  
45.         this.namespace = namespace;  
46.     }  
47.   
48.     public void setZkClient(CuratorFramework zkClient) {  
49.         this.zkClient = zkClient;  
50.     }  
51.   
52.     @Override  
53.     public CuratorFramework getObject() throws Exception {  
54.         if (singleton) {  
55.             if (zkClient == null) {  
56.                 zkClient = create();  
57.                 zkClient.start();  
58.             }  
59.             return zkClient;  
60.         }  
61.         return create();  
62.     }  
63.   
64.     @Override  
65.     public Class<?> getObjectType() {  
66.         return CuratorFramework.class;  
67.     }  
68.   
69.     @Override  
70.     public boolean isSingleton() {  
71.         return singleton;  
72.     }  
73.   
74.     public CuratorFramework create() throws Exception {  
75.         if (StringUtils.isEmpty(namespace)) {  
76.             namespace = ROOT;  
77.         } else {  
78.             namespace = ROOT +"/"+ namespace;  
79.         }  
80.         return create(zkHosts, sessionTimeout, connectionTimeout, namespace);  
81.     }  
82.   
83.     public static CuratorFramework create(String connectString, int sessionTimeout, int connectionTimeout, String namespace) {  
84.         CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder();  
85.         return builder.connectString(connectString).sessionTimeoutMs(sessionTimeout).connectionTimeoutMs(30000)  
86.                 .canBeReadOnly(true).namespace(namespace).retryPolicy(new ExponentialBackoffRetry(1000, Integer.MAX_VALUE))  
87.                 .defaultData(null).build();  
88.     }  
89.   
90.     public void close() {  
91.         if (zkClient != null) {  
92.             zkClient.close();  
93.         }  
94.     }  
95. }  

 

2.服务端注册服务

由于服务端配置需要获取本机的IP地址，因此定义IP获取接口

ThriftServerIpResolve.java

 

1. package cn.slimsmart.thrift.rpc.zookeeper;  
2.   
3. /** 
4.  *  
5.  * 解析thrift-server端IP地址，用于注册服务 
6.  * 1) 可以从一个物理机器或者虚机的特殊文件中解析 
7.  * 2) 可以获取指定网卡序号的Ip 
8.  * 3) 其他 
9.  */  
10. public interface ThriftServerIpResolve {  
11.       
12.     String getServerIp() throws Exception;  
13.       
14.     void reset();  
15.       
16.     //当IP变更时,将会调用reset方法  
17.     static interface IpRestCalllBack{  
18.         public void rest(String newIp);  
19.     }  
20. }  

可以对该接口做不通的实现，下面我们基于网卡获取IP地址，也可以通过配置serverIp

ThriftServerIpLocalNetworkResolve.java

 

 

1. package cn.slimsmart.thrift.rpc.zookeeper;  
2.   
3. import java.net.Inet6Address;  
4. import java.net.InetAddress;  
5. import java.net.NetworkInterface;  
6. import java.net.SocketException;  
7. import java.util.Enumeration;  
8.   
9. import org.slf4j.Logger;  
10. import org.slf4j.LoggerFactory;  
11.   
12. /** 
13.  * 解析网卡Ip 
14.  * 
15.  */  
16. public class ThriftServerIpLocalNetworkResolve implements ThriftServerIpResolve {  
17.       
18.     private Logger logger = LoggerFactory.getLogger(getClass());  
19.   
20.     //缓存  
21.     private String serverIp;  
22.       
23.     public void setServerIp(String serverIp) {  
24.         this.serverIp = serverIp;  
25.     }  
26.   
27.     @Override  
28.     public String getServerIp() {  
29.         if (serverIp != null) {  
30.             return serverIp;  
31.         }  
32.         // 一个主机有多个网络接口  
33.         try {  
34.             Enumeration<NetworkInterface> netInterfaces = NetworkInterface.getNetworkInterfaces();  
35.             while (netInterfaces.hasMoreElements()) {  
36.                 NetworkInterface netInterface = netInterfaces.nextElement();  
37.                 // 每个网络接口,都会有多个"网络地址",比如一定会有lookback地址,会有siteLocal地址等.以及IPV4或者IPV6 .  
38.                 Enumeration<InetAddress> addresses = netInterface.getInetAddresses();  
39.                 while (addresses.hasMoreElements()) {  
40.                     InetAddress address = addresses.nextElement();  
41.                     if(address instanceof Inet6Address){  
42.                         continue;  
43.                     }  
44.                     if (address.isSiteLocalAddress() && !address.isLoopbackAddress()) {  
45.                         serverIp = address.getHostAddress();  
46.                         logger.info("resolve server ip :"+ serverIp);  
47.                         continue;  
48.                     }  
49.                 }  
50.             }  
51.         } catch (SocketException e) {  
52.             e.printStackTrace();  
53.         }  
54.         return serverIp;  
55.     }  
56.   
57.     @Override  
58.     public void reset() {  
59.         serverIp = null;  
60.     }  
61. }  

接下来我们定义发布服务接口，并实现将服务信息（服务接口、版本号，IP、port、weight）发布到zookeeper中。

ThriftServerAddressRegister.java

 

 

1. package cn.slimsmart.thrift.rpc.zookeeper;  
2.   
3. /** 
4.  * 发布服务地址及端口到服务注册中心，这里是zookeeper服务器 
5.  */  
6. public interface ThriftServerAddressRegister {  
7.     /** 
8.      * 发布服务接口 
9.      * @param service 服务接口名称，一个产品中不能重复 
10.      * @param version 服务接口的版本号，默认1.0.0 
11.      * @param address 服务发布的地址和端口 
12.      */  
13.     void register(String service,String version,String address);  
14. }  

实现：ThriftServerAddressRegisterZookeeper.java

 

 

1. package cn.slimsmart.thrift.rpc.zookeeper;  
2.   
3. import java.io.UnsupportedEncodingException;  
4.   
5. import org.apache.curator.framework.CuratorFramework;  
6. import org.apache.curator.framework.imps.CuratorFrameworkState;  
7. import org.apache.zookeeper.CreateMode;  
8. import org.slf4j.Logger;  
9. import org.slf4j.LoggerFactory;  
10. import org.springframework.util.StringUtils;  
11.   
12. import cn.slimsmart.thrift.rpc.ThriftException;  
13.   
14. /** 
15.  *  注册服务列表到Zookeeper 
16.  */  
17. public class ThriftServerAddressRegisterZookeeper implements ThriftServerAddressRegister{  
18.       
19.     private Logger logger = LoggerFactory.getLogger(getClass());  
20.       
21.     private CuratorFramework zkClient;  
22.       
23.     public ThriftServerAddressRegisterZookeeper(){}  
24.       
25.     public ThriftServerAddressRegisterZookeeper(CuratorFramework zkClient){  
26.         this.zkClient = zkClient;  
27.     }  
28.   
29.     public void setZkClient(CuratorFramework zkClient) {  
30.         this.zkClient = zkClient;  
31.     }  
32.   
33.     @Override  
34.     public void register(String service, String version, String address) {  
35.         if(zkClient.getState() == CuratorFrameworkState.LATENT){  
36.             zkClient.start();  
37.         }  
38.         if(StringUtils.isEmpty(version)){  
39.             version="1.0.0";  
40.         }  
41.         //临时节点  
42.         try {  
43.             zkClient.create()  
44.                 .creatingParentsIfNeeded()  
45.                 .withMode(CreateMode.EPHEMERAL)  
46.                 .forPath("/"+service+"/"+version+"/"+address);  
47.         } catch (UnsupportedEncodingException e) {  
48.             logger.error("register service address to zookeeper exception:{}",e);  
49.             throw new ThriftException("register service address to zookeeper exception: address UnsupportedEncodingException", e);  
50.         } catch (Exception e) {  
51.             logger.error("register service address to zookeeper exception:{}",e);  
52.             throw new ThriftException("register service address to zookeeper exception:{}", e);  
53.         }  
54.     }  
55.       
56.     public void close(){  
57.         zkClient.close();  
58.     }  
59. }  

 

 

3.客户端发现服务

定义获取服务地址接口

ThriftServerAddressProvider.java

 

1. package cn.slimsmart.thrift.rpc.zookeeper;  
2.   
3. import java.net.InetSocketAddress;  
4. import java.util.List;  
5.   
6. /** 
7.  * thrift server-service地址提供者,以便构建客户端连接池 
8.  */  
9. public interface ThriftServerAddressProvider {  
10.       
11.     //获取服务名称  
12.     String getService();  
13.   
14.     /** 
15.      * 获取所有服务端地址 
16.      * @return 
17.      */  
18.     List<InetSocketAddress> findServerAddressList();  
19.   
20.     /** 
21.      * 选取一个合适的address,可以随机获取等' 
22.      * 内部可以使用合适的算法. 
23.      * @return 
24.      */  
25.     InetSocketAddress selector();  
26.   
27.     void close();  
28. }  

基于zookeeper服务地址自动发现实现：ThriftServerAddressProviderZookeeper.java

1. package cn.slimsmart.thrift.rpc.zookeeper;  
2.   
3. import java.net.InetSocketAddress;  
4. import java.util.ArrayList;  
5. import java.util.Collections;  
6. import java.util.HashSet;  
7. import java.util.LinkedList;  
8. import java.util.List;  
9. import java.util.Queue;  
10. import java.util.Set;  
11.   
12. import org.apache.curator.framework.CuratorFramework;  
13. import org.apache.curator.framework.imps.CuratorFrameworkState;  
14. import org.apache.curator.framework.recipes.cache.ChildData;  
15. import org.apache.curator.framework.recipes.cache.PathChildrenCache;  
16. import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;  
17. import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;  
18. import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;  
19. import org.slf4j.Logger;  
20. import org.slf4j.LoggerFactory;  
21. import org.springframework.beans.factory.InitializingBean;  
22.   
23. /** 
24.  * 使用zookeeper作为"config"中心,使用apache-curator方法库来简化zookeeper开发 
25.  */  
26. public class ThriftServerAddressProviderZookeeper implements ThriftServerAddressProvider, InitializingBean {  
27.   
28.     private Logger logger = LoggerFactory.getLogger(getClass());  
29.   
30.     // 注册服务  
31.     private String service;  
32.     // 服务版本号  
33.     private String version = "1.0.0";  
34.   
35.     private PathChildrenCache cachedPath;  
36.   
37.     private CuratorFramework zkClient;  
38.   
39.     // 用来保存当前provider所接触过的地址记录  
40.     // 当zookeeper集群故障时,可以使用trace中地址,作为"备份"  
41.     private Set<String> trace = new HashSet<String>();  
42.   
43.     private final List<InetSocketAddress> container = new ArrayList<InetSocketAddress>();  
44.   
45.     private Queue<InetSocketAddress> inner = new LinkedList<InetSocketAddress>();  
46.   
47.     private Object lock = new Object();  
48.   
49.     // 默认权重  
50.     private static final Integer DEFAULT_WEIGHT = 1;  
51.   
52.     public void setService(String service) {  
53.         this.service = service;  
54.     }  
55.   
56.     public void setVersion(String version) {  
57.         this.version = version;  
58.     }  
59.   
60.     public ThriftServerAddressProviderZookeeper() {  
61.     }  
62.   
63.     public ThriftServerAddressProviderZookeeper(CuratorFramework zkClient) {  
64.         this.zkClient = zkClient;  
65.     }  
66.   
67.     public void setZkClient(CuratorFramework zkClient) {  
68.         this.zkClient = zkClient;  
69.     }  
70.   
71.     @Override  
72.     public void afterPropertiesSet() throws Exception {  
73.         // 如果zk尚未启动,则启动  
74.         if (zkClient.getState() == CuratorFrameworkState.LATENT) {  
75.             zkClient.start();  
76.         }  
77.         buildPathChildrenCache(zkClient, getServicePath(), true);  
78.         cachedPath.start(StartMode.POST_INITIALIZED_EVENT);  
79.     }  
80.   
81.     private String getServicePath(){  
82.         return "/" + service + "/" + version;  
83.     }  
84.     private void buildPathChildrenCache(final CuratorFramework client, String path, Boolean cacheData) throws Exception {  
85.         cachedPath = new PathChildrenCache(client, path, cacheData);  
86.         cachedPath.getListenable().addListener(new PathChildrenCacheListener() {  
87.             @Override  
88.             public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {  
89.                 PathChildrenCacheEvent.Type eventType = event.getType();  
90.                 switch (eventType) {  
91.                 case CONNECTION_RECONNECTED:  
92.                     logger.info("Connection is reconection.");  
93.                     break;  
94.                 case CONNECTION_SUSPENDED:  
95.                     logger.info("Connection is suspended.");  
96.                     break;  
97.                 case CONNECTION_LOST:  
98.                     logger.warn("Connection error,waiting...");  
99.                     return;  
100.                 default:  
101.                     //  
102.                 }  
103.                 // 任何节点的时机数据变动,都会rebuild,此处为一个"简单的"做法.  
104.                 cachedPath.rebuild();  
105.                 rebuild();  
106.             }  
107.   
108.             protected void rebuild() throws Exception {  
109.                 List<ChildData> children = cachedPath.getCurrentData();  
110.                 if (children == null || children.isEmpty()) {  
111.                     // 有可能所有的thrift server都与zookeeper断开了链接  
112.                     // 但是,有可能,thrift client与thrift server之间的网络是良好的  
113.                     // 因此此处是否需要清空container,是需要多方面考虑的.  
114.                     container.clear();  
115.                     logger.error("thrift server-cluster error....");  
116.                     return;  
117.                 }  
118.                 List<InetSocketAddress> current = new ArrayList<InetSocketAddress>();  
119.                 String path = null;  
120.                 for (ChildData data : children) {  
121.                     path = data.getPath();  
122.                     logger.debug("get path:"+path);  
123.                     path = path.substring(getServicePath().length()+1);  
124.                     logger.debug("get serviceAddress:"+path);  
125.                     String address = new String(path.getBytes(), "utf-8");  
126.                     current.addAll(transfer(address));  
127.                     trace.add(address);  
128.                 }  
129.                 Collections.shuffle(current);  
130.                 synchronized (lock) {  
131.                     container.clear();  
132.                     container.addAll(current);  
133.                     inner.clear();  
134.                     inner.addAll(current);  
135.   
136.                 }  
137.             }  
138.         });  
139.     }  
140.   
141.     private List<InetSocketAddress> transfer(String address) {  
142.         String[] hostname = address.split(":");  
143.         Integer weight = DEFAULT_WEIGHT;  
144.         if (hostname.length == 3) {  
145.             weight = Integer.valueOf(hostname[2]);  
146.         }  
147.         String ip = hostname[0];  
148.         Integer port = Integer.valueOf(hostname[1]);  
149.         List<InetSocketAddress> result = new ArrayList<InetSocketAddress>();  
150.         // 根据优先级，将ip：port添加多次到地址集中，然后随机取地址实现负载  
151.         for (int i = 0; i < weight; i++) {  
152.             result.add(new InetSocketAddress(ip, port));  
153.         }  
154.         return result;  
155.     }  
156.   
157.     @Override  
158.     public List<InetSocketAddress> findServerAddressList() {  
159.         return Collections.unmodifiableList(container);  
160.     }  
161.   
162.     @Override  
163.     public synchronized InetSocketAddress selector() {  
164.         if (inner.isEmpty()) {  
165.             if (!container.isEmpty()) {  
166.                 inner.addAll(container);  
167.             } else if (!trace.isEmpty()) {  
168.                 synchronized (lock) {  
169.                     for (String hostname : trace) {  
170.                         container.addAll(transfer(hostname));  
171.                     }  
172.                     Collections.shuffle(container);  
173.                     inner.addAll(container);  
174.                 }  
175.             }  
176.         }  
177.         return inner.poll();  
178.     }  
179.   
180.     @Override  
181.     public void close() {  
182.         try {  
183.             cachedPath.close();  
184.             zkClient.close();  
185.         } catch (Exception e) {  
186.         }  
187.     }  
188.   
189.     @Override  
190.     public String getService() {  
191.         return service;  
192.     }  
193.   
194. }  

对此接口还做了一种实现，通过配置获取服务地址，参考附件：FixedAddressProvider.java

 

三、服务端服务注册实现

ThriftServiceServerFactory.java

 

1. package cn.slimsmart.thrift.rpc;  
2.   
3. import java.lang.instrument.IllegalClassFormatException;  
4. import java.lang.reflect.Constructor;  
5.   
6. import org.apache.thrift.TProcessor;  
7. import org.apache.thrift.TProcessorFactory;  
8. import org.apache.thrift.protocol.TBinaryProtocol;  
9. import org.apache.thrift.server.TServer;  
10. import org.apache.thrift.server.TThreadedSelectorServer;  
11. import org.apache.thrift.transport.TFramedTransport;  
12. import org.apache.thrift.transport.TNonblockingServerSocket;  
13. import org.springframework.beans.factory.InitializingBean;  
14. import org.springframework.util.StringUtils;  
15.   
16. import cn.slimsmart.thrift.rpc.zookeeper.ThriftServerAddressRegister;  
17. import cn.slimsmart.thrift.rpc.zookeeper.ThriftServerIpLocalNetworkResolve;  
18. import cn.slimsmart.thrift.rpc.zookeeper.ThriftServerIpResolve;  
19.   
20. /** 
21.  * 服务端注册服务工厂 
22.  */  
23. public class ThriftServiceServerFactory implements InitializingBean {  
24.     // 服务注册本机端口  
25.     private Integer port = 8299;  
26.     // 优先级  
27.     private Integer weight = 1;// default  
28.     // 服务实现类  
29.     private Object service;// serice实现类  
30.     //服务版本号  
31.     private String version;  
32.     // 解析本机IP  
33.     private ThriftServerIpResolve thriftServerIpResolve;  
34.     //服务注册  
35.     private ThriftServerAddressRegister thriftServerAddressRegister;  
36.   
37.     private ServerThread serverThread;  
38.       
39.     public void setPort(Integer port) {  
40.         this.port = port;  
41.     }  
42.   
43.     public void setWeight(Integer weight) {  
44.         this.weight = weight;  
45.     }  
46.   
47.     public void setService(Object service) {  
48.         this.service = service;  
49.     }  
50.   
51.     public void setVersion(String version) {  
52.         this.version = version;  
53.     }  
54.   
55.     public void setThriftServerIpResolve(ThriftServerIpResolve thriftServerIpResolve) {  
56.         this.thriftServerIpResolve = thriftServerIpResolve;  
57.     }  
58.   
59.     public void setThriftServerAddressRegister(ThriftServerAddressRegister thriftServerAddressRegister) {  
60.         this.thriftServerAddressRegister = thriftServerAddressRegister;  
61.     }  
62.   
63.     @Override  
64.     public void afterPropertiesSet() throws Exception {  
65.         if (thriftServerIpResolve == null) {  
66.             thriftServerIpResolve = new ThriftServerIpLocalNetworkResolve();  
67.         }  
68.         String serverIP = thriftServerIpResolve.getServerIp();  
69.         if (StringUtils.isEmpty(serverIP)) {  
70.             throw new ThriftException("cant find server ip...");  
71.         }  
72.   
73.         String hostname = serverIP + ":" + port + ":" + weight;  
74.         Class<?> serviceClass = service.getClass();  
75.         // 获取实现类接口  
76.         Class<?>[] interfaces = serviceClass.getInterfaces();  
77.         if (interfaces.length == 0) {  
78.             throw new IllegalClassFormatException("service-class should implements Iface");  
79.         }  
80.         // reflect,load "Processor";  
81.         TProcessor processor = null;  
82.         String serviceName = null;  
83.         for (Class<?> clazz : interfaces) {  
84.             String cname = clazz.getSimpleName();  
85.             if (!cname.equals("Iface")) {  
86.                 continue;  
87.             }  
88.             serviceName = clazz.getEnclosingClass().getName();  
89.             String pname = serviceName + "$Processor";  
90.             try {  
91.                 ClassLoader classLoader = Thread.currentThread().getContextClassLoader();  
92.                 Class<?> pclass = classLoader.loadClass(pname);  
93.                 if (!TProcessor.class.isAssignableFrom(pclass)) {  
94.                     continue;  
95.                 }  
96.                 Constructor<?> constructor = pclass.getConstructor(clazz);  
97.                 processor = (TProcessor) constructor.newInstance(service);  
98.                 break;  
99.             } catch (Exception e) {  
100.                 //  
101.             }  
102.         }  
103.         if (processor == null) {  
104.             throw new IllegalClassFormatException("service-class should implements Iface");  
105.         }  
106.         //需要单独的线程,因为serve方法是阻塞的.  
107.         serverThread = new ServerThread(processor, port);  
108.         serverThread.start();  
109.         // 注册服务  
110.         if (thriftServerAddressRegister != null) {  
111.             thriftServerAddressRegister.register(serviceName, version, hostname);  
112.         }  
113.   
114.     }  
115.     class ServerThread extends Thread {  
116.         private TServer server;  
117.         ServerThread(TProcessor processor, int port) throws Exception {  
118.             TNonblockingServerSocket serverTransport = new TNonblockingServerSocket(port);  
119.             TThreadedSelectorServer.Args tArgs = new TThreadedSelectorServer.Args(serverTransport);    
120.             TProcessorFactory processorFactory = new TProcessorFactory(processor);  
121.             tArgs.processorFactory(processorFactory);  
122.             tArgs.transportFactory(new TFramedTransport.Factory());    
123.             tArgs.protocolFactory( new TBinaryProtocol.Factory(true, true));   
124.             server = new TThreadedSelectorServer(tArgs);  
125.         }  
126.   
127.         @Override  
128.         public void run(){  
129.             try{  
130.                 //启动服务  
131.                 server.serve();  
132.             }catch(Exception e){  
133.                 //  
134.             }  
135.         }  
136.           
137.         public void stopServer(){  
138.             server.stop();  
139.         }  
140.     }  
141.       
142.     public void close() {  
143.         serverThread.stopServer();  
144.     }  
145. }  

 

 

四、客户端获取服务代理及连接池实现

客户端连接池实现：ThriftClientPoolFactory.java

 

1. package cn.slimsmart.thrift.rpc;  
2.   
3. import java.net.InetSocketAddress;  
4.   
5. import org.apache.commons.pool.BasePoolableObjectFactory;  
6. import org.apache.thrift.TServiceClient;  
7. import org.apache.thrift.TServiceClientFactory;  
8. import org.apache.thrift.protocol.TBinaryProtocol;  
9. import org.apache.thrift.protocol.TProtocol;  
10. import org.apache.thrift.transport.TFramedTransport;  
11. import org.apache.thrift.transport.TSocket;  
12. import org.apache.thrift.transport.TTransport;  
13.   
14. import cn.slimsmart.thrift.rpc.zookeeper.ThriftServerAddressProvider;  
15.   
16. /** 
17.  * 连接池,thrift-client for spring 
18.  */  
19. public class ThriftClientPoolFactory extends BasePoolableObjectFactory<TServiceClient> {  
20.   
21.     private final ThriftServerAddressProvider serverAddressProvider;  
22.     private final TServiceClientFactory<TServiceClient> clientFactory;  
23.     private PoolOperationCallBack callback;  
24.   
25.     protected ThriftClientPoolFactory(ThriftServerAddressProvider addressProvider, TServiceClientFactory<TServiceClient> clientFactory) throws Exception {  
26.         this.serverAddressProvider = addressProvider;  
27.         this.clientFactory = clientFactory;  
28.     }  
29.   
30.     protected ThriftClientPoolFactory(ThriftServerAddressProvider addressProvider, TServiceClientFactory<TServiceClient> clientFactory,  
31.             PoolOperationCallBack callback) throws Exception {  
32.         this.serverAddressProvider = addressProvider;  
33.         this.clientFactory = clientFactory;  
34.         this.callback = callback;  
35.     }  
36.   
37.     static interface PoolOperationCallBack {  
38.         // 销毁client之前执行  
39.         void destroy(TServiceClient client);  
40.   
41.         // 创建成功是执行  
42.         void make(TServiceClient client);  
43.     }  
44.   
45.     public void destroyObject(TServiceClient client) throws Exception {  
46.         if (callback != null) {  
47.             try {  
48.                 callback.destroy(client);  
49.             } catch (Exception e) {  
50.                 //  
51.             }  
52.         }  
53.         TTransport pin = client.getInputProtocol().getTransport();  
54.         pin.close();  
55.     }  
56.   
57.     public boolean validateObject(TServiceClient client) {  
58.         TTransport pin = client.getInputProtocol().getTransport();  
59.         return pin.isOpen();  
60.     }  
61.   
62.     @Override  
63.     public TServiceClient makeObject() throws Exception {  
64.         InetSocketAddress address = serverAddressProvider.selector();  
65.         TSocket tsocket = new TSocket(address.getHostName(), address.getPort());  
66.         TTransport transport = new TFramedTransport(tsocket);  
67.         TProtocol protocol = new TBinaryProtocol(transport);  
68.         TServiceClient client = this.clientFactory.getClient(protocol);  
69.         transport.open();  
70.         if (callback != null) {  
71.             try {  
72.                 callback.make(client);  
73.             } catch (Exception e) {  
74.                 //  
75.             }  
76.         }  
77.         return client;  
78.     }  
79.   
80. }  

客户端服务代理工厂实现：ThriftServiceClientProxyFactory.java

 

 

1. package cn.slimsmart.thrift.rpc;  
2.   
3. import java.lang.reflect.InvocationHandler;  
4. import java.lang.reflect.Method;  
5. import java.lang.reflect.Proxy;  
6.   
7. import org.apache.commons.pool.impl.GenericObjectPool;  
8. import org.apache.thrift.TServiceClient;  
9. import org.apache.thrift.TServiceClientFactory;  
10. import org.springframework.beans.factory.FactoryBean;  
11. import org.springframework.beans.factory.InitializingBean;  
12.   
13. import cn.slimsmart.thrift.rpc.ThriftClientPoolFactory.PoolOperationCallBack;  
14. import cn.slimsmart.thrift.rpc.zookeeper.ThriftServerAddressProvider;  
15.   
16. /** 
17.  * 客户端代理 
18.  */  
19. @SuppressWarnings({ "unchecked", "rawtypes" })  
20. public class ThriftServiceClientProxyFactory implements FactoryBean, InitializingBean {  
21.   
22.     private Integer maxActive = 32;// 最大活跃连接数  
23.   
24.     // ms,default 3 min,链接空闲时间  
25.     // -1,关闭空闲检测  
26.     private Integer idleTime = 180000;  
27.     private ThriftServerAddressProvider serverAddressProvider;  
28.   
29.     private Object proxyClient;  
30.     private Class<?> objectClass;  
31.   
32.     private GenericObjectPool<TServiceClient> pool;  
33.   
34.     private PoolOperationCallBack callback = new PoolOperationCallBack() {  
35.         @Override  
36.         public void make(TServiceClient client) {  
37.             System.out.println("create");  
38.         }  
39.   
40.         @Override  
41.         public void destroy(TServiceClient client) {  
42.             System.out.println("destroy");  
43.         }  
44.     };  
45.       
46.     public void setMaxActive(Integer maxActive) {  
47.         this.maxActive = maxActive;  
48.     }  
49.   
50.     public void setIdleTime(Integer idleTime) {  
51.         this.idleTime = idleTime;  
52.     }  
53.   
54.     public void setServerAddressProvider(ThriftServerAddressProvider serverAddressProvider) {  
55.         this.serverAddressProvider = serverAddressProvider;  
56.     }  
57.   
58.     @Override  
59.     public void afterPropertiesSet() throws Exception {  
60.         ClassLoader classLoader = Thread.currentThread().getContextClassLoader();  
61.         // 加载Iface接口  
62.         objectClass = classLoader.loadClass(serverAddressProvider.getService() + "$Iface");  
63.         // 加载Client.Factory类  
64.         Class<TServiceClientFactory<TServiceClient>> fi = (Class<TServiceClientFactory<TServiceClient>>) classLoader.loadClass(serverAddressProvider.getService() + "$Client$Factory");  
65.         TServiceClientFactory<TServiceClient> clientFactory = fi.newInstance();  
66.         ThriftClientPoolFactory clientPool = new ThriftClientPoolFactory(serverAddressProvider, clientFactory, callback);  
67.         GenericObjectPool.Config poolConfig = new GenericObjectPool.Config();  
68.         poolConfig.maxActive = maxActive;  
69.         poolConfig.minIdle = 0;  
70.         poolConfig.minEvictableIdleTimeMillis = idleTime;  
71.         poolConfig.timeBetweenEvictionRunsMillis = idleTime / 2L;  
72.         pool = new GenericObjectPool<TServiceClient>(clientPool, poolConfig);  
73.         proxyClient = Proxy.newProxyInstance(classLoader, new Class[] { objectClass }, new InvocationHandler() {  
74.             @Override  
75.             public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {  
76.                 //  
77.                 TServiceClient client = pool.borrowObject();  
78.                 try {  
79.                     return method.invoke(client, args);  
80.                 } catch (Exception e) {  
81.                     throw e;  
82.                 } finally {  
83.                     pool.returnObject(client);  
84.                 }  
85.             }  
86.         });  
87.     }  
88.   
89.     @Override  
90.     public Object getObject() throws Exception {  
91.         return proxyClient;  
92.     }  
93.   
94.     @Override  
95.     public Class<?> getObjectType() {  
96.         return objectClass;  
97.     }  
98.   
99.     @Override  
100.     public boolean isSingleton() {  
101.         return true;  
102.     }  
103.   
104.     public void close() {  
105.         if (serverAddressProvider != null) {  
106.             serverAddressProvider.close();  
107.         }  
108.     }  
109. }  

下面我们看一下服务端和客户端的配置;

 

服务端spring-context-thrift-server.xml

 

1. <?xml version="1.0" encoding="UTF-8"?>  
2. <beans xmlns=""  
3.     xmlns:xsi="" xmlns:context=""  
4.     xmlns:aop="" xmlns:tx=""  
5.     xsi:schemaLocation="   
6.                    
7.                    
8.                  "  
9.     default-lazy-init="false">  
10.   
11.     <!-- zookeeper -->  
12.     <bean id="thriftZookeeper" class="cn.slimsmart.thrift.rpc.zookeeper.ZookeeperFactory"  
13.         destroy-method="close">  
14.         <property name="zkHosts"  
15.             value="192.168.36.54:2181,192.168.36.99:2181,192.168.36.189:2181" />  
16.         <property name="namespace" value="cn.slimsmart.thrift.rpc.demo" />  
17.         <property name="connectionTimeout" value="3000" />  
18.         <property name="sessionTimeout" value="3000" />  
19.         <property name="singleton" value="true" />  
20.     </bean>  
21.     <bean id="sericeAddressRegister"  
22.         class="cn.slimsmart.thrift.rpc.zookeeper.ThriftServerAddressRegisterZookeeper"  
23.         destroy-method="close">  
24.         <property name="zkClient" ref="thriftZookeeper" />  
25.     </bean>  
26.     <bean id="echoSerivceImpl" class="cn.slimsmart.thrift.rpc.demo.EchoSerivceImpl" />  
27.   
28.     <bean id="echoSerivce" class="cn.slimsmart.thrift.rpc.ThriftServiceServerFactory"  
29.         destroy-method="close">  
30.         <property name="service" ref="echoSerivceImpl" />  
31.         <property name="port" value="9000" />  
32.         <property name="version" value="1.0.0" />  
33.         <property name="weight" value="1" />  
34.         <property name="thriftServerAddressRegister" ref="sericeAddressRegister" />  
35.     </bean>  
36.       
37.     <bean id="echoSerivce1" class="cn.slimsmart.thrift.rpc.ThriftServiceServerFactory"  
38.         destroy-method="close">  
39.         <property name="service" ref="echoSerivceImpl" />  
40.         <property name="port" value="9001" />  
41.         <property name="version" value="1.0.0" />  
42.         <property name="weight" value="1" />  
43.         <property name="thriftServerAddressRegister" ref="sericeAddressRegister" />  
44.     </bean>  
45.       
46.     <bean id="echoSerivce2" class="cn.slimsmart.thrift.rpc.ThriftServiceServerFactory"  
47.         destroy-method="close">  
48.         <property name="service" ref="echoSerivceImpl" />  
49.         <property name="port" value="9002" />  
50.         <property name="version" value="1.0.0" />  
51.         <property name="weight" value="1" />  
52.         <property name="thriftServerAddressRegister" ref="sericeAddressRegister" />  
53.     </bean>  
54. </beans>  

客户端：spring-context-thrift-client.xml

1. <?xml version="1.0" encoding="UTF-8"?>  
2. <beans xmlns=""  
3.     xmlns:xsi="" xmlns:context=""  
4.     xmlns:aop="" xmlns:tx=""  
5.     xsi:schemaLocation="   
6.                    
7.                    
8.                  "  
9.     default-lazy-init="false">  
10.       
11.     <!-- fixedAddress -->  
12.     <!--   
13.     <bean id="fixedAddressProvider" class="cn.slimsmart.thrift.rpc.zookeeper.FixedAddressProvider">  
14.          <property name="service" value="cn.slimsmart.thrift.rpc.demo.EchoSerivce" />  
15.          <property name="serverAddress" value="192.168.36.215:9001:1,192.168.36.215:9002:2,192.168.36.215:9003:3" />  
16.     </bean>  
17.     <bean id="echoSerivce" class="cn.slimsmart.thrift.rpc.ThriftServiceClientProxyFactory">  
18.         <property name="maxActive" value="5" />  
19.         <property name="idleTime" value="10000" />  
20.         <property name="serverAddressProvider" ref="fixedAddressProvider" />  
21.     </bean>  
22.    -->  
23.     <!-- zookeeper   -->  
24.     <bean id="thriftZookeeper" class="cn.slimsmart.thrift.rpc.zookeeper.ZookeeperFactory"  
25.         destroy-method="close">  
26.         <property name="zkHosts"  
27.             value="192.168.36.54:2181,192.168.36.99:2181,192.168.36.189:2181" />  
28.         <property name="namespace" value="cn.slimsmart.thrift.rpc.demo" />  
29.         <property name="connectionTimeout" value="3000" />  
30.         <property name="sessionTimeout" value="3000" />  
31.         <property name="singleton" value="true" />  
32.     </bean>  
33.     <bean id="echoSerivce" class="cn.slimsmart.thrift.rpc.ThriftServiceClientProxyFactory" destroy-method="close">  
34.         <property name="maxActive" value="5" />  
35.         <property name="idleTime" value="1800000" />  
36.         <property name="serverAddressProvider">  
37.             <bean class="cn.slimsmart.thrift.rpc.zookeeper.ThriftServerAddressProviderZookeeper">  
38.                 <property name="service" value="cn.slimsmart.thrift.rpc.demo.EchoSerivce" />  
39.                 <property name="version" value="1.0.0" />  
40.                 <property name="zkClient" ref="thriftZookeeper" />  
41.             </bean>  
42.         </property>  
43.     </bean>  
44. </beans>  

 

运行服务端后，我们可以看见zookeeper注册了多个服务地址。

 

问题

安装文档：

遇到一个bison版本过低的问题：

 

参考资料

Thrift原理简析（Java）－

Thrift－

Thrift与ZooKeeper－

Thrift专题－

Thrift服务端与Spring－

Thrift客户端与Spring－