zoukankan      html  css  js  c++  java

  • 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方式的通信：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 (schemeField.id) {  
    9.           case 1: // ID  
    10.             if (schemeField.type == org.apache.thrift.protocol.TType.I32) {  
    11.               struct.id = 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="http://www.springframework.org/schema/beans"  
    3.     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"  
    4.     xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"  
    5.     xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd  
    6.                 http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.0.xsd  
    7.                 http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.0.xsd  
    8.                 http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.0.xsd"  
    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="http://www.springframework.org/schema/beans"  
    3.     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"  
    4.     xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"  
    5.     xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd  
    6.                 http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.0.xsd  
    7.                 http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.0.xsd  
    8.                 http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.0.xsd"  
    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版本过低的问题：http://blog.csdn.net/qq910894904/article/details/41132779
     
    参考资料
    Thrift原理简析（Java）－http://shift-alt-ctrl.iteye.com/blog/1987416
     
  • 相关阅读:
    httpclient + TestNG 接口自动测试 第二章
    httpclient + TestNG 接口自动测试 第一章
    Kafka-manager启动命令
    多台服务器搭建Spark集群
    Scala学习 -- 基础语法
    Spark学习 -- RDD
    Spark 学习
    Angular constructor和OnInit的区别和适用场景
    TypeScript基础学习 —— 变量声明
    TypeScript基础学习 —— 基础类型
  • 原文地址：https://www.cnblogs.com/scott19820130/p/4940274.html
Copyright © 2011-2022 走看看