Thrift全面介紹

本文轉載自查看原文 2015-11-05 17:36 1787 Java/ 開源框架/ Thrift

官網： http://thrift.apache.org

簡介

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

struct User {
1:i64 id,
2:string name,
3:i64 timestamp,
4:bool vip
}
service UserService {
User getById(1:i64 id)
}

你可以將自己的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代碼

//windows平台下,將API文件輸出在service目錄下(此目錄需要存在)
> thrift.exe --gen java -o service service.thrift

需要明確的是：Thrift和其他RPC框架不同，thrift在生成的API文件中，已經描述了"調用過程"(即硬編碼)，而不是像其他RPC那樣在運行時(runtime)動態解析方法調用或者參數。

UserService實現類

Java代碼

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

實現類，需要放在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代碼

//read方法逐個讀取字段,按照"索引",最終將"struct"對象封裝完畢.
//write方法也非常類似,按照"索引"順序逐個輸出到流中.
while (true){
schemeField = iprot.readFieldBegin();
if (schemeField.type == org.apache.thrift.protocol.TType.STOP) {
break;
}
switch (schemeField.id) {
case 1: // ID
if (schemeField.type == org.apache.thrift.protocol.TType.I32) {
struct.id = iprot.readI32();
struct.setIdIsSet(true);
} else {
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
break;
case 2: // NAME
..
}
}

因為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代碼

//參見:TServiceClient
//API方法調用時,發送請求數據流
protected void sendBase(String methodName, TBase args) throws TException {
oprot_.writeMessageBegin(new TMessage(methodName, TMessageType.CALL, ++seqid_));//首先寫入"方法名稱"和"seqid_"
args.write(oprot_);//序列化參數
oprot_.writeMessageEnd();
oprot_.getTransport().flush();
}
protected void receiveBase(TBase result, String methodName) throws TException {
TMessage msg = iprot_.readMessageBegin();//如果執行有異常
if (msg.type == TMessageType.EXCEPTION) {
TApplicationException x = TApplicationException.read(iprot_);
iprot_.readMessageEnd();
throw x;
}//檢測seqid是否一致
if (msg.seqid != seqid_) {
throw new TApplicationException(TApplicationException.BAD_SEQUENCE_ID, methodName + " failed: out of sequence response");
}
result.read(iprot_);//反序列化
iprot_.readMessageEnd();
}

Thrift提供了簡單的容錯方式：每次方法調用，都會在Client端標記一個seqid，這是一個自增的本地ID，在TCP請求時將此seqid追加到流中，同時Server端響應時，也將此seqid原樣返回過來；這樣客戶端就可以根據此值用來判斷"請求--響應"是對應的，如果出現亂序，將會導致此請求以異常的方式結束。

2) 響應

Java代碼

//參考: TBaseProcessor.java
@Override
public boolean process(TProtocol in, TProtocol out) throws TException {
TMessage msg = in.readMessageBegin();
ProcessFunction fn = processMap.get(msg.name);//根據方法名,查找"內部類"
if (fn == null) {
TProtocolUtil.skip(in, TType.STRUCT);
in.readMessageEnd();
TApplicationException x = new TApplicationException(TApplicationException.UNKNOWN_METHOD, "Invalid method name: '"+msg.name+"'");
out.writeMessageBegin(new TMessage(msg.name, TMessageType.EXCEPTION, msg.seqid));
x.write(out);//序列化響應結果,直接輸出
out.writeMessageEnd();
out.getTransport().flush();
return true;
}
fn.process(msg.seqid, in, out, iface);
return true;
}

thrift生成的UserService.Processor類,就是server端用來處理請求過程的"代理類";server端從socket中讀取請求需要調用的"方法名" +參數列表,並交付給Processor類處理;和其他的RPC調用不同的時,thrift並沒有使用類似於"反射機制"的方式來調用方法,而是將UserService的每個方法生成一個"內部類":

Java代碼

public static class getById extends org.apache.thrift.ProcessFunction<I, getById_args> {
public getById() {
super("getById");//其中getById為標識符
}
public getById_args getEmptyArgsInstance() {
return new getById_args();
}
protected boolean isOneway() {
return false;
}
//實際處理方法
public getById_result getResult(I iface, getById_args args) throws org.apache.thrift.TException {
getById_result result = new getById_result();
result.success = iface.getById(args.id);
return result;
}
}

這個"內部類",將會在Processor初始化的時候,放入到一個map中,此后即可以通過"方法名"查找,然后調用其"getResult"方法了.

Java代碼

public static class Processor extends org.apache.thrift.TBaseProcessor implements org.apache.thrift.TProcessor {
public Processor(I iface) {
super(iface, getProcessMap(new HashMap<String, org.apache.thrift.ProcessFunction<I, ? extends org.apache.thrift.TBase>>()));
}
protected Processor(I iface, Map<String, org.apache.thrift.ProcessFunction<I, ? extends org.apache.thrift.TBase>> processMap) {
super(iface, getProcessMap(processMap));
}
private static 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) {
//放入map
processMap.put("getById", new getById());
return processMap;
}
....
}

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

Java代碼

TThreadPoolServer
public void serve() {
try {
//啟動服務
serverTransport_.listen();
} catch (TTransportException ttx) {
LOGGER.error("Error occurred during listening.", ttx);
return;
}
// Run the preServe event
if (eventHandler_ != null) {
eventHandler_.preServe();
}
stopped_ = false;
setServing(true);
//循環,直到被關閉
while (!stopped_) {
int failureCount = 0;
try {
//accept客戶端Socket鏈接,
//對於每個新鏈接,將會封裝成runnable,並提交給線程或者線程池中運行.
TTransport client = serverTransport_.accept();
WorkerProcess wp = new WorkerProcess(client);
executorService_.execute(wp);
} catch (TTransportException ttx) {
if (!stopped_) {
++failureCount;
LOGGER.warn("Transport error occurred during acceptance of message.", ttx);
}
}
}
//....
}

Thrift Server端，設計思路也非常的直接。當前Service server啟動之后，將會以阻塞的方式偵聽Socket鏈接(代碼參考TThreadPoolServer)，每建立一個Socket鏈接，都會將此Socket經過封裝之后，放入線程池中，本質上也是一個Socket鏈接對應一個Worker Thread。這個Thread只會處理此Socket中的所有數據請求，直到Socket關閉。

Java代碼

//參考:WorkerProcess
while (true) {
if (eventHandler != null) {
eventHandler.processContext(connectionContext, inputTransport, outputTransport);
}
if(stopped_ || !processor.process(inputProtocol, outputProtocol)) {
break;
}
}

當有Socket鏈接不是很多的時候，TThreadPoolServer並不會有太大的性能問題，可以通過指定ThreadPool中線程的個數進行簡單的調優。如果Socket鏈接很多，我們只能使用TThreadedSelectorServer來做支撐，TThreadedSelectorServer內部基於NIO模式，具有異步的特性，可以極大的提升server端的並發能力；不過在絕大多數情況下，在thrift中使用"異步"似乎不太容易讓人接受，畢竟這意味着Client端需要阻塞，並且在高並發環境中這個阻塞時間是不可控的。但SelecorServer確實可以有效的提升Server的並發能力，而且在一定程度上可以提升吞吐能力，這或許是我們優化Thrift Server比較可靠的方式之一。

Client端代碼示例

Java代碼

public class UserServiceClient {
public void startClient() {
TTransport transport;
try {
transport = new TSocket("localhost", 1234);
TProtocol protocol = new TBinaryProtocol(transport);
UserService.Client client = new UserService.Client(protocol);
transport.open();
User user = client.getById(1000);
////
transport.close();
} catch (TTransportException e) {
e.printStackTrace();
} catch (TException e) {
e.printStackTrace();
}
}
}

Server端代碼示例

Java代碼

public class Server {
public void startServer() {
try {
TServerSocket serverTransport = new TServerSocket(1234);
UserService.Processor process = new Processor(new UserServiceImpl());
Factory portFactory = new TBinaryProtocol.Factory(true, true);
Args args = new Args(serverTransport);
args.processor(process);
args.protocolFactory(portFactory);
TServer server = new TThreadPoolServer(args);
server.serve();
} catch (TTransportException e) {
e.printStackTrace();
}
}
}

到這里，你就會發現，一個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

<dependencies>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>3.0.7.RELEASE</version>
</dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.5</version>
</dependency>
<!--
<dependency>
<groupId>com.101tec</groupId>
<artifactId>zkclient</artifactId>
<version>0.4</version>
</dependency>
-->
<dependency>
<groupId>org.apache.thrift</groupId>
<artifactId>libthrift</artifactId>
<version>0.9.1</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>2.3.0</version>
</dependency>
<dependency>
<groupId>commons-pool</groupId>
<artifactId>commons-pool</artifactId>
<version>1.6</version>
</dependency>
</dependencies>

本實例,使用了apache-curator作為zookeeper客戶端.

2. spring-thrift-server.xml

<bean id="thriftZookeeper" class="com.demo.thrift.zookeeper.ZookeeperFactory" destroy-method="close">
<property name="connectString" value="127.0.0.1:2181"></property>
<property name="namespace" value="demo/thrift-service"></property>
</bean>
<bean id="sericeAddressReporter" class="com.demo.thrift.support.impl.DynamicAddressReporter" destroy-method="close">
<property name="zookeeper" ref="thriftZookeeper"></property>
</bean>
<bean id="userService" class="com.demo.service.UserServiceImpl"/>
<bean class="com.demo.thrift.ThriftServiceServerFactory" destroy-method="close">
<property name="service" ref="userService"></property>
<property name="configPath" value="UserServiceImpl"></property>
<property name="port" value="9090"></property>
<property name="addressReporter" ref="sericeAddressReporter"></property>
</bean>

3. ThriftServiceServerFactory.java

此類嚴格上說並不是一個工廠類,它的主要作用就是封裝指定的"service" ,然后啟動一個server的過程,其中"service"屬性表示服務的實現類,addressReporter表示當server啟動成功后,需要指定的操作(比如,向zookeeper發送service的IP信息).

究竟當前server的ip地址是多少,在不同的設計中,有所不同,比如:有些管理員喜歡將本機的IP地址寫入到os下的某個文件中,如果上層應用需要獲取可靠的IP信息,就需要讀取這個文件...你可以實現自己的ThriftServerIpTransfer來獲取當前server的IP.

為了減少xml中的配置信息,在factory中,使用了反射機制來構建"Processor"類.

public class ThriftServiceServerFactory implements InitializingBean {
private Integer port;
private Integer priority = 1;// default
private Object service;// serice實現類
private ThriftServerIpTransfer ipTransfer;
private ThriftServerAddressReporter addressReporter;
private ServerThread serverThread;
private String configPath;
public void setService(Object service) {
this.service = service;
}
public void setPriority(Integer priority) {
this.priority = priority;
}
public void setPort(Integer port) {
this.port = port;
}
public void setIpTransfer(ThriftServerIpTransfer ipTransfer) {
this.ipTransfer = ipTransfer;
}
public void setAddressReporter(ThriftServerAddressReporter addressReporter) {
this.addressReporter = addressReporter;
}
public void setConfigPath(String configPath) {
this.configPath = configPath;
}
@Override
public void afterPropertiesSet() throws Exception {
if (ipTransfer == null) {
ipTransfer = new LocalNetworkIpTransfer();
}
String ip = ipTransfer.getIp();
if (ip == null) {
throw new NullPointerException("cant find server ip...");
}
String hostname = ip + ":" + port + ":" + priority;
Class serviceClass = service.getClass();
ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
Class<?>[] interfaces = serviceClass.getInterfaces();
if (interfaces.length == 0) {
throw new IllegalClassFormatException("service-class should implements Iface");
}
// reflect,load "Processor";
Processor processor = null;
for (Class clazz : interfaces) {
String cname = clazz.getSimpleName();
if (!cname.equals("Iface")) {
continue;
}
String pname = clazz.getEnclosingClass().getName() + "$Processor";
try {
Class pclass = classLoader.loadClass(pname);
if (!pclass.isAssignableFrom(Processor.class)) {
continue;
}
Constructor constructor = pclass.getConstructor(clazz);
processor = (Processor) constructor.newInstance(service);
break;
} catch (Exception e) {
//
}
}
if (processor == null) {
throw new IllegalClassFormatException("service-class should implements Iface");
}
//需要單獨的線程,因為serve方法是阻塞的.
serverThread = new ServerThread(processor, port);
serverThread.start();
// report
if (addressReporter != null) {
addressReporter.report(configPath, hostname);
}
}
class ServerThread extends Thread {
private TServer server;
ServerThread(Processor processor, int port) throws Exception {
TServerSocket serverTransport = new TServerSocket(port);
Factory portFactory = new TBinaryProtocol.Factory(true, true);
Args args = new Args(serverTransport);
args.processor(processor);
args.protocolFactory(portFactory);
server = new TThreadPoolServer(args);
}
@Override
public void run(){
try{
server.serve();
}catch(Exception e){
//
}
}
public void stopServer(){
server.stop();
}
}
public void close() {
serverThread.stopServer();
}
}

4. DynamicAddressReporter.java

在ThriftServiceServerFactory中,有個可選的屬性:addressReporter, DynamicAddressReporter提供了向zookeeper注冊service信息的能力,當server啟動正常后,把server的IP + port發送到zookeeper中;那么此后服務消費client,就可以從zookeeper中獲取server列表,並與它們建立鏈接(池).這樣client端只需要關注zookeeper的節點名稱即可,不需要配置大量的ip+port.

public class DynamicAddressReporter implements ThriftServerAddressReporter {
private CuratorFramework zookeeper;
public DynamicAddressReporter(){}
public DynamicAddressReporter(CuratorFramework zookeeper){
this.zookeeper = zookeeper;
}
public void setZookeeper(CuratorFramework zookeeper) {
this.zookeeper = zookeeper;
}
@Override
public void report(String service, String address) throws Exception {
if(zookeeper.getState() == CuratorFrameworkState.LATENT){
zookeeper.start();
zookeeper.newNamespaceAwareEnsurePath(service);
}
zookeeper.create()
.creatingParentsIfNeeded()
.withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
.forPath(service +"/i_",address.getBytes("utf-8"));
}
public void close(){
zookeeper.close();
}
}

5. 測試類

public class ServiceMain {
/**
* @param args
*/
public static void main(String[] args) {
try {
ApplicationContext context = new ClassPathXmlApplicationContext("spring-thrift-server.xml");
Thread.sleep(3000000);
} catch (Exception e) {
e.printStackTrace();
}
}
}

客戶端開發

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

<dependencies>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>3.0.7.RELEASE</version>
</dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.5</version>
</dependency>
<!--
<dependency>
<groupId>com.101tec</groupId>
<artifactId>zkclient</artifactId>
<version>0.4</version>
</dependency>
-->
<dependency>
<groupId>org.apache.thrift</groupId>
<artifactId>libthrift</artifactId>
<version>0.9.1</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>2.3.0</version>
</dependency>
<dependency>
<groupId>commons-pool</groupId>
<artifactId>commons-pool</artifactId>
<version>1.6</version>
</dependency>
</dependencies>

2. spring-thrift-client.xml

其中zookeeper作為可選項,開發者也可以通過制定serverAddress的方式指定server的地址.

<!--
<bean id="userService" class="com.demo.thrift.ThriftServiceClientProxyFactory">
<property name="service" value="com.demo.service.UserService"></property>
<property name="serverAddress" value="127.0.0.1:9090:2"></property>
<property name="maxActive" value="5"></property>
<property name="idleTime" value="10000"></property>
</bean>
-->
<bean id="thriftZookeeper" class="com.demo.thrift.zookeeper.ZookeeperFactory" destroy-method="close">
<property name="connectString" value="127.0.0.1:2181"></property>
<property name="namespace" value="demo/thrift-service"></property>
</bean>
<bean id="userService" class="com.demo.thrift.ThriftServiceClientProxyFactory" destroy-method="close">
<property name="service" value="com.demo.service.UserService"></property>
<property name="maxActive" value="5"></property>
<property name="idleTime" value="1800000"></property>
<property name="addressProvider">
<bean class="com.demo.thrift.support.impl.DynamicAddressProvider">
<property name="configPath" value="UserServiceImpl"></property>
<property name="zookeeper" ref="thriftZookeeper"></property>
</bean>
</property>
</bean>

3. ThriftServiceClientProxyFactory.java

因為我們要在client端使用連接池方案,那么就需要對client的方法調用過程,進行代理,這個類,就是維護了一個"Client"代理類,並在方法調用時,從"對象池"中取出一個"Client"對象,並在方法實際調用結束后歸還給"對象池".

@SuppressWarnings("rawtypes")
public class ThriftServiceClientProxyFactory implements FactoryBean,InitializingBean {
private String service;
private String serverAddress;
private Integer maxActive = 32;//最大活躍連接數
////ms,default 3 min,鏈接空閑時間
//-1,關閉空閑檢測
private Integer idleTime = 180000;
private ThriftServerAddressProvider addressProvider;
private Object proxyClient;
public void setMaxActive(Integer maxActive) {
this.maxActive = maxActive;
}
public void setIdleTime(Integer idleTime) {
this.idleTime = idleTime;
}
public void setService(String service) {
this.service = service;
}
public void setServerAddress(String serverAddress) {
this.serverAddress = serverAddress;
}
public void setAddressProvider(ThriftServerAddressProvider addressProvider) {
this.addressProvider = addressProvider;
}
private Class objectClass;
private GenericObjectPool<TServiceClient> pool;
private PoolOperationCallBack callback = new PoolOperationCallBack() {
@Override
public void make(TServiceClient client) {
System.out.println("create");
}
@Override
public void destroy(TServiceClient client) {
System.out.println("destroy");
}
};
@Override
public void afterPropertiesSet() throws Exception {
if(serverAddress != null){
addressProvider = new FixedAddressProvider(serverAddress);
}
ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
//加載Iface接口
objectClass = classLoader.loadClass(service + "$Iface");
//加載Client.Factory類
Class<TServiceClientFactory<TServiceClient>> fi = (Class<TServiceClientFactory<TServiceClient>>)classLoader.loadClass(service + "$Client$Factory");
TServiceClientFactory<TServiceClient> clientFactory = fi.newInstance();
ThriftClientPoolFactory clientPool = new ThriftClientPoolFactory(addressProvider, clientFactory,callback);
GenericObjectPool.Config poolConfig = new GenericObjectPool.Config();
poolConfig.maxActive = maxActive;
poolConfig.minIdle = 0;
poolConfig.minEvictableIdleTimeMillis = idleTime;
poolConfig.timeBetweenEvictionRunsMillis = idleTime/2L;
pool = new GenericObjectPool<TServiceClient>(clientPool,poolConfig);
proxyClient = Proxy.newProxyInstance(classLoader,new Class[]{objectClass},new InvocationHandler() {
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
//
TServiceClient client = pool.borrowObject();
try{
return method.invoke(client, args);
}catch(Exception e){
throw e;
}finally{
pool.returnObject(client);
}
}
});
}
@Override
public Object getObject() throws Exception {
return proxyClient;
}
@Override
public Class<?> getObjectType() {
return objectClass;
}
@Override
public boolean isSingleton() {
return true; //To change body of implemented methods use File | Settings | File Templates.
}
public void close(){
if(addressProvider != null){
addressProvider.close();
}
}
}

4. ThriftClientPoolFactory.java

"Client"對象池,對象池中是已經實例化的Client對象,Client對象負責與Thrift server通信.

/**
* 連接池,thrift-client for spring
*/
public class ThriftClientPoolFactory extends BasePoolableObjectFactory<TServiceClient>{
private final ThriftServerAddressProvider addressProvider;
private final TServiceClientFactory<TServiceClient> clientFactory;
private PoolOperationCallBack callback;
protected ThriftClientPoolFactory(ThriftServerAddressProvider addressProvider,TServiceClientFactory<TServiceClient> clientFactory) throws Exception {
this.addressProvider = addressProvider;
this.clientFactory = clientFactory;
}
protected ThriftClientPoolFactory(ThriftServerAddressProvider addressProvider,TServiceClientFactory<TServiceClient> clientFactory,PoolOperationCallBack callback) throws Exception {
this.addressProvider = addressProvider;
this.clientFactory = clientFactory;
this.callback = callback;
}
@Override
public TServiceClient makeObject() throws Exception {
InetSocketAddress address = addressProvider.selector();
TSocket tsocket = new TSocket(address.getHostName(),address.getPort());
TProtocol protocol = new TBinaryProtocol(tsocket);
TServiceClient client = this.clientFactory.getClient(protocol);
tsocket.open();
if(callback != null){
try{
callback.make(client);
}catch(Exception e){
//
}
}
return client;
}
public void destroyObject(TServiceClient client) throws Exception {
if(callback != null){
try{
callback.destroy(client);
}catch(Exception e){
//
}
}
TTransport pin = client.getInputProtocol().getTransport();
pin.close();
}
public boolean validateObject(TServiceClient client) {
TTransport pin = client.getInputProtocol().getTransport();
return pin.isOpen();
}
static interface PoolOperationCallBack {
//銷毀client之前執行
void destroy(TServiceClient client);
//創建成功是執行
void make(TServiceClient client);
}
}

5. DynamicAddressProvider.java

將zookeeper作為server地址的提供者,這樣客戶端就不需要再配置文件中指定一堆ip + port,而且當server服務有更新時,也不需要client端重新配置.

/**
* 可以動態獲取address地址,方案設計參考
* 1) 可以間歇性的調用一個web-service來獲取地址
* 2) 可以使用事件監聽機制,被動的接收消息,來獲取最新的地址(比如基於MQ,nio等)
* 3) 可以基於zookeeper-watcher機制,獲取最新地址
*
* 本實例,使用zookeeper作為"config"中心,使用apache-curator方法庫來簡化zookeeper開發
* 如下實現,僅供參考
*/
public class DynamicAddressProvider implements ThriftServerAddressProvider, InitializingBean {
private String configPath;
private PathChildrenCache cachedPath;
private CuratorFramework zookeeper;
//用來保存當前provider所接觸過的地址記錄
//當zookeeper集群故障時,可以使用trace中地址,作為"備份"
private Set<String> trace = new HashSet<String>();
private final List<InetSocketAddress> container = new ArrayList<InetSocketAddress>();
private Queue<InetSocketAddress> inner = new LinkedList<InetSocketAddress>();
private Object lock = new Object();
private static final Integer DEFAULT_PRIORITY = 1;
public void setConfigPath(String configPath) {
this.configPath = configPath;
}
public void setZookeeper(CuratorFramework zookeeper) {
this.zookeeper = zookeeper;
}
@Override
public void afterPropertiesSet() throws Exception {
//如果zk尚未啟動,則啟動
if(zookeeper.getState() == CuratorFrameworkState.LATENT){
zookeeper.start();
}
buildPathChildrenCache(zookeeper, configPath, true);
cachedPath.start(StartMode.POST_INITIALIZED_EVENT);
}
private void buildPathChildrenCache(CuratorFramework client, String path, Boolean cacheData) throws Exception {
cachedPath = new PathChildrenCache(client, path, cacheData);
cachedPath.getListenable().addListener(new PathChildrenCacheListener() {
@Override
public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
PathChildrenCacheEvent.Type eventType = event.getType();
switch (eventType) {
// case CONNECTION_RECONNECTED:
//
// break;
case CONNECTION_SUSPENDED:
case CONNECTION_LOST:
System.out.println("Connection error,waiting...");
return;
default:
//
}
//任何節點的時機數據變動,都會rebuild,此處為一個"簡單的"做法.
cachedPath.rebuild();
rebuild();
}
protected void rebuild() throws Exception {
List<ChildData> children = cachedPath.getCurrentData();
if (children == null || children.isEmpty()) {
//有可能所有的thrift server都與zookeeper斷開了鏈接
//但是,有可能,thrift client與thrift server之間的網絡是良好的
//因此此處是否需要清空container,是需要多方面考慮的.
container.clear();
System.out.println("thrift server-cluster error....");
return;
}
List<InetSocketAddress> current = new ArrayList<InetSocketAddress>();
for (ChildData data : children) {
String address = new String(data.getData(), "utf-8");
current.addAll(transfer(address));
trace.add(address);
}
Collections.shuffle(current);
synchronized (lock) {
container.clear();
container.addAll(current);
inner.clear();
inner.addAll(current);
}
}
});
}
private List<InetSocketAddress> transfer(String address){
String[] hostname = address.split(":");
Integer priority = DEFAULT_PRIORITY;
if (hostname.length == 3) {
priority = Integer.valueOf(hostname[2]);
}
String ip = hostname[0];
Integer port = Integer.valueOf(hostname[1]);
List<InetSocketAddress> result = new ArrayList<InetSocketAddress>();
for (int i = 0; i < priority; i++) {
result.add(new InetSocketAddress(ip, port));
}
return result;
}
@Override
public List<InetSocketAddress> getAll() {
return Collections.unmodifiableList(container);
}
@Override
public synchronized InetSocketAddress selector() {
if (inner.isEmpty()) {
if(!container.isEmpty()){
inner.addAll(container);
}else if(!trace.isEmpty()){
synchronized (lock) {
for(String hostname : trace){
container.addAll(transfer(hostname));
}
Collections.shuffle(container);
inner.addAll(container);
}
}
}
return inner.poll();//null
}
@Override
public void close() {
try {
cachedPath.close();
zookeeper.close();
} catch (Exception e) {
//
}
}
}

到此為止，我們的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

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

Client API的方式, 不推薦, 我們推薦Service接口的方式(服務化)。

下面我們來一一實現：

一、pom.xml引入依賴jar包

<dependency>
<groupId>org.apache.thrift</groupId>
<artifactId>libthrift</artifactId>
<version>0.9.2</version>
</dependency>
<dependency>
<groupId>commons-pool</groupId>
<artifactId>commons-pool</artifactId>
<version>1.6</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>4.0.9.RELEASE</version>
</dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.6</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>2.7.1</version>
</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;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.util.StringUtils;
/**
* 獲取zookeeper客戶端鏈接
*/
public class ZookeeperFactory implements FactoryBean<CuratorFramework> {
private String zkHosts;
// session超時
private int sessionTimeout = 30000;
private int connectionTimeout = 30000;
// 共享一個zk鏈接
private boolean singleton = true;
// 全局path前綴,常用來區分不同的應用
private String namespace;
private final static String ROOT = "rpc";
private CuratorFramework zkClient;
public void setZkHosts(String zkHosts) {
this.zkHosts = zkHosts;
}
public void setSessionTimeout(int sessionTimeout) {
this.sessionTimeout = sessionTimeout;
}
public void setConnectionTimeout(int connectionTimeout) {
this.connectionTimeout = connectionTimeout;
}
public void setSingleton(boolean singleton) {
this.singleton = singleton;
}
public void setNamespace(String namespace) {
this.namespace = namespace;
}
public void setZkClient(CuratorFramework zkClient) {
this.zkClient = zkClient;
}
@Override
public CuratorFramework getObject() throws Exception {
if (singleton) {
if (zkClient == null) {
zkClient = create();
zkClient.start();
}
return zkClient;
}
return create();
}
@Override
public Class<?> getObjectType() {
return CuratorFramework.class;
}
@Override
public boolean isSingleton() {
return singleton;
}
public CuratorFramework create() throws Exception {
if (StringUtils.isEmpty(namespace)) {
namespace = ROOT;
} else {
namespace = ROOT +"/"+ namespace;
}
return create(zkHosts, sessionTimeout, connectionTimeout, namespace);
}
public static CuratorFramework create(String connectString, int sessionTimeout, int connectionTimeout, String namespace) {
CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder();
return builder.connectString(connectString).sessionTimeoutMs(sessionTimeout).connectionTimeoutMs(30000)
.canBeReadOnly(true).namespace(namespace).retryPolicy(new ExponentialBackoffRetry(1000, Integer.MAX_VALUE))
.defaultData(null).build();
}
public void close() {
if (zkClient != null) {
zkClient.close();
}
}
}

2.服務端注冊服務

由於服務端配置需要獲取本機的IP地址，因此定義IP獲取接口

ThriftServerIpResolve.java

package cn.slimsmart.thrift.rpc.zookeeper;
/**
*
* 解析thrift-server端IP地址，用於注冊服務
* 1) 可以從一個物理機器或者虛機的特殊文件中解析
* 2) 可以獲取指定網卡序號的Ip
* 3) 其他
*/
public interface ThriftServerIpResolve {
String getServerIp() throws Exception;
void reset();
//當IP變更時,將會調用reset方法
static interface IpRestCalllBack{
public void rest(String newIp);
}
}

可以對該接口做不通的實現，下面我們基於網卡獲取IP地址，也可以通過配置serverIp
ThriftServerIpLocalNetworkResolve.java

package cn.slimsmart.thrift.rpc.zookeeper;
import java.net.Inet6Address;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.util.Enumeration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* 解析網卡Ip
*
*/
public class ThriftServerIpLocalNetworkResolve implements ThriftServerIpResolve {
private Logger logger = LoggerFactory.getLogger(getClass());
//緩存
private String serverIp;
public void setServerIp(String serverIp) {
this.serverIp = serverIp;
}
@Override
public String getServerIp() {
if (serverIp != null) {
return serverIp;
}
// 一個主機有多個網絡接口
try {
Enumeration<NetworkInterface> netInterfaces = NetworkInterface.getNetworkInterfaces();
while (netInterfaces.hasMoreElements()) {
NetworkInterface netInterface = netInterfaces.nextElement();
// 每個網絡接口,都會有多個"網絡地址",比如一定會有lookback地址,會有siteLocal地址等.以及IPV4或者IPV6 .
Enumeration<InetAddress> addresses = netInterface.getInetAddresses();
while (addresses.hasMoreElements()) {
InetAddress address = addresses.nextElement();
if(address instanceof Inet6Address){
continue;
}
if (address.isSiteLocalAddress() && !address.isLoopbackAddress()) {
serverIp = address.getHostAddress();
logger.info("resolve server ip :"+ serverIp);
continue;
}
}
}
} catch (SocketException e) {
e.printStackTrace();
}
return serverIp;
}
@Override
public void reset() {
serverIp = null;
}
}

接下來我們定義發布服務接口，並實現將服務信息（服務接口、版本號，IP、port、weight）發布到zookeeper中。

ThriftServerAddressRegister.java

package cn.slimsmart.thrift.rpc.zookeeper;
/**
* 發布服務地址及端口到服務注冊中心，這里是zookeeper服務器
*/
public interface ThriftServerAddressRegister {
/**
* 發布服務接口
* @param service 服務接口名稱，一個產品中不能重復
* @param version 服務接口的版本號，默認1.0.0
* @param address 服務發布的地址和端口
*/
void register(String service,String version,String address);
}

實現：ThriftServerAddressRegisterZookeeper.java

package cn.slimsmart.thrift.rpc.zookeeper;
import java.io.UnsupportedEncodingException;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;
import cn.slimsmart.thrift.rpc.ThriftException;
/**
* 注冊服務列表到Zookeeper
*/
public class ThriftServerAddressRegisterZookeeper implements ThriftServerAddressRegister{
private Logger logger = LoggerFactory.getLogger(getClass());
private CuratorFramework zkClient;
public ThriftServerAddressRegisterZookeeper(){}
public ThriftServerAddressRegisterZookeeper(CuratorFramework zkClient){
this.zkClient = zkClient;
}
public void setZkClient(CuratorFramework zkClient) {
this.zkClient = zkClient;
}
@Override
public void register(String service, String version, String address) {
if(zkClient.getState() == CuratorFrameworkState.LATENT){
zkClient.start();
}
if(StringUtils.isEmpty(version)){
version="1.0.0";
}
//臨時節點
try {
zkClient.create()
.creatingParentsIfNeeded()
.withMode(CreateMode.EPHEMERAL)
.forPath("/"+service+"/"+version+"/"+address);
} catch (UnsupportedEncodingException e) {
logger.error("register service address to zookeeper exception:{}",e);
throw new ThriftException("register service address to zookeeper exception: address UnsupportedEncodingException", e);
} catch (Exception e) {
logger.error("register service address to zookeeper exception:{}",e);
throw new ThriftException("register service address to zookeeper exception:{}", e);
}
}
public void close(){
zkClient.close();
}
}

3.客戶端發現服務

定義獲取服務地址接口

ThriftServerAddressProvider.java

package cn.slimsmart.thrift.rpc.zookeeper;
import java.net.InetSocketAddress;
import java.util.List;
/**
* thrift server-service地址提供者,以便構建客戶端連接池
*/
public interface ThriftServerAddressProvider {
//獲取服務名稱
String getService();
/**
* 獲取所有服務端地址
* @return
*/
List<InetSocketAddress> findServerAddressList();
/**
* 選取一個合適的address,可以隨機獲取等'
* 內部可以使用合適的算法.
* @return
*/
InetSocketAddress selector();
void close();
}

基於zookeeper服務地址自動發現實現：ThriftServerAddressProviderZookeeper.java

package cn.slimsmart.thrift.rpc.zookeeper;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.Set;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
/**
* 使用zookeeper作為"config"中心,使用apache-curator方法庫來簡化zookeeper開發
*/
public class ThriftServerAddressProviderZookeeper implements ThriftServerAddressProvider, InitializingBean {
private Logger logger = LoggerFactory.getLogger(getClass());
// 注冊服務
private String service;
// 服務版本號
private String version = "1.0.0";
private PathChildrenCache cachedPath;
private CuratorFramework zkClient;
// 用來保存當前provider所接觸過的地址記錄
// 當zookeeper集群故障時,可以使用trace中地址,作為"備份"
private Set<String> trace = new HashSet<String>();
private final List<InetSocketAddress> container = new ArrayList<InetSocketAddress>();
private Queue<InetSocketAddress> inner = new LinkedList<InetSocketAddress>();
private Object lock = new Object();
// 默認權重
private static final Integer DEFAULT_WEIGHT = 1;
public void setService(String service) {
this.service = service;
}
public void setVersion(String version) {
this.version = version;
}
public ThriftServerAddressProviderZookeeper() {
}
public ThriftServerAddressProviderZookeeper(CuratorFramework zkClient) {
this.zkClient = zkClient;
}
public void setZkClient(CuratorFramework zkClient) {
this.zkClient = zkClient;
}
@Override
public void afterPropertiesSet() throws Exception {
// 如果zk尚未啟動,則啟動
if (zkClient.getState() == CuratorFrameworkState.LATENT) {
zkClient.start();
}
buildPathChildrenCache(zkClient, getServicePath(), true);
cachedPath.start(StartMode.POST_INITIALIZED_EVENT);
}
private String getServicePath(){
return "/" + service + "/" + version;
}
private void buildPathChildrenCache(final CuratorFramework client, String path, Boolean cacheData) throws Exception {
cachedPath = new PathChildrenCache(client, path, cacheData);
cachedPath.getListenable().addListener(new PathChildrenCacheListener() {
@Override
public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
PathChildrenCacheEvent.Type eventType = event.getType();
switch (eventType) {
case CONNECTION_RECONNECTED:
logger.info("Connection is reconection.");
break;
case CONNECTION_SUSPENDED:
logger.info("Connection is suspended.");
break;
case CONNECTION_LOST:
logger.warn("Connection error,waiting...");
return;
default:
//
}
// 任何節點的時機數據變動,都會rebuild,此處為一個"簡單的"做法.
cachedPath.rebuild();
rebuild();
}
protected void rebuild() throws Exception {
List<ChildData> children = cachedPath.getCurrentData();
if (children == null || children.isEmpty()) {
// 有可能所有的thrift server都與zookeeper斷開了鏈接
// 但是,有可能,thrift client與thrift server之間的網絡是良好的
// 因此此處是否需要清空container,是需要多方面考慮的.
container.clear();
logger.error("thrift server-cluster error....");
return;
}
List<InetSocketAddress> current = new ArrayList<InetSocketAddress>();
String path = null;
for (ChildData data : children) {
path = data.getPath();
logger.debug("get path:"+path);
path = path.substring(getServicePath().length()+1);
logger.debug("get serviceAddress:"+path);
String address = new String(path.getBytes(), "utf-8");
current.addAll(transfer(address));
trace.add(address);
}
Collections.shuffle(current);
synchronized (lock) {
container.clear();
container.addAll(current);
inner.clear();
inner.addAll(current);
}
}
});
}
private List<InetSocketAddress> transfer(String address) {
String[] hostname = address.split(":");
Integer weight = DEFAULT_WEIGHT;
if (hostname.length == 3) {
weight = Integer.valueOf(hostname[2]);
}
String ip = hostname[0];
Integer port = Integer.valueOf(hostname[1]);
List<InetSocketAddress> result = new ArrayList<InetSocketAddress>();
// 根據優先級，將ip：port添加多次到地址集中，然后隨機取地址實現負載
for (int i = 0; i < weight; i++) {
result.add(new InetSocketAddress(ip, port));
}
return result;
}
@Override
public List<InetSocketAddress> findServerAddressList() {
return Collections.unmodifiableList(container);
}
@Override
public synchronized InetSocketAddress selector() {
if (inner.isEmpty()) {
if (!container.isEmpty()) {
inner.addAll(container);
} else if (!trace.isEmpty()) {
synchronized (lock) {
for (String hostname : trace) {
container.addAll(transfer(hostname));
}
Collections.shuffle(container);
inner.addAll(container);
}
}
}
return inner.poll();
}
@Override
public void close() {
try {
cachedPath.close();
zkClient.close();
} catch (Exception e) {
}
}
@Override
public String getService() {
return service;
}
}

對此接口還做了一種實現，通過配置獲取服務地址，參考附件：FixedAddressProvider.java

三、服務端服務注冊實現

ThriftServiceServerFactory.java

package cn.slimsmart.thrift.rpc;
import java.lang.instrument.IllegalClassFormatException;
import java.lang.reflect.Constructor;
import org.apache.thrift.TProcessor;
import org.apache.thrift.TProcessorFactory;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.server.TServer;
import org.apache.thrift.server.TThreadedSelectorServer;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TNonblockingServerSocket;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.StringUtils;
import cn.slimsmart.thrift.rpc.zookeeper.ThriftServerAddressRegister;
import cn.slimsmart.thrift.rpc.zookeeper.ThriftServerIpLocalNetworkResolve;
import cn.slimsmart.thrift.rpc.zookeeper.ThriftServerIpResolve;
/**
* 服務端注冊服務工廠
*/
public class ThriftServiceServerFactory implements InitializingBean {
// 服務注冊本機端口
private Integer port = 8299;
// 優先級
private Integer weight = 1;// default
// 服務實現類
private Object service;// serice實現類
//服務版本號
private String version;
// 解析本機IP
private ThriftServerIpResolve thriftServerIpResolve;
//服務注冊
private ThriftServerAddressRegister thriftServerAddressRegister;
private ServerThread serverThread;
public void setPort(Integer port) {
this.port = port;
}
public void setWeight(Integer weight) {
this.weight = weight;
}
public void setService(Object service) {
this.service = service;
}
public void setVersion(String version) {
this.version = version;
}
public void setThriftServerIpResolve(ThriftServerIpResolve thriftServerIpResolve) {
this.thriftServerIpResolve = thriftServerIpResolve;
}
public void setThriftServerAddressRegister(ThriftServerAddressRegister thriftServerAddressRegister) {
this.thriftServerAddressRegister = thriftServerAddressRegister;
}
@Override
public void afterPropertiesSet() throws Exception {
if (thriftServerIpResolve == null) {
thriftServerIpResolve = new ThriftServerIpLocalNetworkResolve();
}
String serverIP = thriftServerIpResolve.getServerIp();
if (StringUtils.isEmpty(serverIP)) {
throw new ThriftException("cant find server ip...");
}
String hostname = serverIP + ":" + port + ":" + weight;
Class<?> serviceClass = service.getClass();
// 獲取實現類接口
Class<?>[] interfaces = serviceClass.getInterfaces();
if (interfaces.length == 0) {
throw new IllegalClassFormatException("service-class should implements Iface");
}
// reflect,load "Processor";
TProcessor processor = null;
String serviceName = null;
for (Class<?> clazz : interfaces) {
String cname = clazz.getSimpleName();
if (!cname.equals("Iface")) {
continue;
}
serviceName = clazz.getEnclosingClass().getName();
String pname = serviceName + "$Processor";
try {
ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
Class<?> pclass = classLoader.loadClass(pname);
if (!TProcessor.class.isAssignableFrom(pclass)) {
continue;
}
Constructor<?> constructor = pclass.getConstructor(clazz);
processor = (TProcessor) constructor.newInstance(service);
break;
} catch (Exception e) {
//
}
}
if (processor == null) {
throw new IllegalClassFormatException("service-class should implements Iface");
}
//需要單獨的線程,因為serve方法是阻塞的.
serverThread = new ServerThread(processor, port);
serverThread.start();
// 注冊服務
if (thriftServerAddressRegister != null) {
thriftServerAddressRegister.register(serviceName, version, hostname);
}
}
class ServerThread extends Thread {
private TServer server;
ServerThread(TProcessor processor, int port) throws Exception {
TNonblockingServerSocket serverTransport = new TNonblockingServerSocket(port);
TThreadedSelectorServer.Args tArgs = new TThreadedSelectorServer.Args(serverTransport);
TProcessorFactory processorFactory = new TProcessorFactory(processor);
tArgs.processorFactory(processorFactory);
tArgs.transportFactory(new TFramedTransport.Factory());
tArgs.protocolFactory( new TBinaryProtocol.Factory(true, true));
server = new TThreadedSelectorServer(tArgs);
}
@Override
public void run(){
try{
//啟動服務
server.serve();
}catch(Exception e){
//
}
}
public void stopServer(){
server.stop();
}
}
public void close() {
serverThread.stopServer();
}
}

四、客戶端獲取服務代理及連接池實現
客戶端連接池實現：ThriftClientPoolFactory.java

package cn.slimsmart.thrift.rpc;
import java.net.InetSocketAddress;
import org.apache.commons.pool.BasePoolableObjectFactory;
import org.apache.thrift.TServiceClient;
import org.apache.thrift.TServiceClientFactory;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import cn.slimsmart.thrift.rpc.zookeeper.ThriftServerAddressProvider;
/**
* 連接池,thrift-client for spring
*/
public class ThriftClientPoolFactory extends BasePoolableObjectFactory<TServiceClient> {
private final ThriftServerAddressProvider serverAddressProvider;
private final TServiceClientFactory<TServiceClient> clientFactory;
private PoolOperationCallBack callback;
protected ThriftClientPoolFactory(ThriftServerAddressProvider addressProvider, TServiceClientFactory<TServiceClient> clientFactory) throws Exception {
this.serverAddressProvider = addressProvider;
this.clientFactory = clientFactory;
}
protected ThriftClientPoolFactory(ThriftServerAddressProvider addressProvider, TServiceClientFactory<TServiceClient> clientFactory,
PoolOperationCallBack callback) throws Exception {
this.serverAddressProvider = addressProvider;
this.clientFactory = clientFactory;
this.callback = callback;
}
static interface PoolOperationCallBack {
// 銷毀client之前執行
void destroy(TServiceClient client);
// 創建成功是執行
void make(TServiceClient client);
}
public void destroyObject(TServiceClient client) throws Exception {
if (callback != null) {
try {
callback.destroy(client);
} catch (Exception e) {
//
}
}
TTransport pin = client.getInputProtocol().getTransport();
pin.close();
}
public boolean validateObject(TServiceClient client) {
TTransport pin = client.getInputProtocol().getTransport();
return pin.isOpen();
}
@Override
public TServiceClient makeObject() throws Exception {
InetSocketAddress address = serverAddressProvider.selector();
TSocket tsocket = new TSocket(address.getHostName(), address.getPort());
TTransport transport = new TFramedTransport(tsocket);
TProtocol protocol = new TBinaryProtocol(transport);
TServiceClient client = this.clientFactory.getClient(protocol);
transport.open();
if (callback != null) {
try {
callback.make(client);
} catch (Exception e) {
//
}
}
return client;
}
}

客戶端服務代理工廠實現：ThriftServiceClientProxyFactory.java

package cn.slimsmart.thrift.rpc;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import org.apache.commons.pool.impl.GenericObjectPool;
import org.apache.thrift.TServiceClient;
import org.apache.thrift.TServiceClientFactory;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import cn.slimsmart.thrift.rpc.ThriftClientPoolFactory.PoolOperationCallBack;
import cn.slimsmart.thrift.rpc.zookeeper.ThriftServerAddressProvider;
/**
* 客戶端代理
*/
@SuppressWarnings({ "unchecked", "rawtypes" })
public class ThriftServiceClientProxyFactory implements FactoryBean, InitializingBean {
private Integer maxActive = 32;// 最大活躍連接數
// ms,default 3 min,鏈接空閑時間
// -1,關閉空閑檢測
private Integer idleTime = 180000;
private ThriftServerAddressProvider serverAddressProvider;
private Object proxyClient;
private Class<?> objectClass;
private GenericObjectPool<TServiceClient> pool;
private PoolOperationCallBack callback = new PoolOperationCallBack() {
@Override
public void make(TServiceClient client) {
System.out.println("create");
}
@Override
public void destroy(TServiceClient client) {
System.out.println("destroy");
}
};
public void setMaxActive(Integer maxActive) {
this.maxActive = maxActive;
}
public void setIdleTime(Integer idleTime) {
this.idleTime = idleTime;
}
public void setServerAddressProvider(ThriftServerAddressProvider serverAddressProvider) {
this.serverAddressProvider = serverAddressProvider;
}
@Override
public void afterPropertiesSet() throws Exception {
ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
// 加載Iface接口
objectClass = classLoader.loadClass(serverAddressProvider.getService() + "$Iface");
// 加載Client.Factory類
Class<TServiceClientFactory<TServiceClient>> fi = (Class<TServiceClientFactory<TServiceClient>>) classLoader.loadClass(serverAddressProvider.getService() + "$Client$Factory");
TServiceClientFactory<TServiceClient> clientFactory = fi.newInstance();
ThriftClientPoolFactory clientPool = new ThriftClientPoolFactory(serverAddressProvider, clientFactory, callback);
GenericObjectPool.Config poolConfig = new GenericObjectPool.Config();
poolConfig.maxActive = maxActive;
poolConfig.minIdle = 0;
poolConfig.minEvictableIdleTimeMillis = idleTime;
poolConfig.timeBetweenEvictionRunsMillis = idleTime / 2L;
pool = new GenericObjectPool<TServiceClient>(clientPool, poolConfig);
proxyClient = Proxy.newProxyInstance(classLoader, new Class[] { objectClass }, new InvocationHandler() {
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
//
TServiceClient client = pool.borrowObject();
try {
return method.invoke(client, args);
} catch (Exception e) {
throw e;
} finally {
pool.returnObject(client);
}
}
});
}
@Override
public Object getObject() throws Exception {
return proxyClient;
}
@Override
public Class<?> getObjectType() {
return objectClass;
}
@Override
public boolean isSingleton() {
return true;
}
public void close() {
if (serverAddressProvider != null) {
serverAddressProvider.close();
}
}
}

下面我們看一下服務端和客戶端的配置;

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

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.0.xsd
http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.0.xsd
http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.0.xsd"
default-lazy-init="false">
<bean id="thriftZookeeper" class="cn.slimsmart.thrift.rpc.zookeeper.ZookeeperFactory"
destroy-method="close">
<property name="zkHosts"
value="192.168.36.54:2181,192.168.36.99:2181,192.168.36.189:2181" />
<property name="namespace" value="cn.slimsmart.thrift.rpc.demo" />
<property name="connectionTimeout" value="3000" />
<property name="sessionTimeout" value="3000" />
<property name="singleton" value="true" />
</bean>
<bean id="sericeAddressRegister"
class="cn.slimsmart.thrift.rpc.zookeeper.ThriftServerAddressRegisterZookeeper"
destroy-method="close">
<property name="zkClient" ref="thriftZookeeper" />
</bean>
<bean id="echoSerivceImpl" class="cn.slimsmart.thrift.rpc.demo.EchoSerivceImpl" />
<bean id="echoSerivce" class="cn.slimsmart.thrift.rpc.ThriftServiceServerFactory"
destroy-method="close">
<property name="service" ref="echoSerivceImpl" />
<property name="port" value="9000" />
<property name="version" value="1.0.0" />
<property name="weight" value="1" />
<property name="thriftServerAddressRegister" ref="sericeAddressRegister" />
</bean>
<bean id="echoSerivce1" class="cn.slimsmart.thrift.rpc.ThriftServiceServerFactory"
destroy-method="close">
<property name="service" ref="echoSerivceImpl" />
<property name="port" value="9001" />
<property name="version" value="1.0.0" />
<property name="weight" value="1" />
<property name="thriftServerAddressRegister" ref="sericeAddressRegister" />
</bean>
<bean id="echoSerivce2" class="cn.slimsmart.thrift.rpc.ThriftServiceServerFactory"
destroy-method="close">
<property name="service" ref="echoSerivceImpl" />
<property name="port" value="9002" />
<property name="version" value="1.0.0" />
<property name="weight" value="1" />
<property name="thriftServerAddressRegister" ref="sericeAddressRegister" />
</bean>
</beans>

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

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.0.xsd
http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.0.xsd
http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.0.xsd"
default-lazy-init="false">
<!--
<bean id="fixedAddressProvider" class="cn.slimsmart.thrift.rpc.zookeeper.FixedAddressProvider">
<property name="service" value="cn.slimsmart.thrift.rpc.demo.EchoSerivce" />
<property name="serverAddress" value="192.168.36.215:9001:1,192.168.36.215:9002:2,192.168.36.215:9003:3" />
</bean>
<bean id="echoSerivce" class="cn.slimsmart.thrift.rpc.ThriftServiceClientProxyFactory">
<property name="maxActive" value="5" />
<property name="idleTime" value="10000" />
<property name="serverAddressProvider" ref="fixedAddressProvider" />
</bean>
-->
<bean id="thriftZookeeper" class="cn.slimsmart.thrift.rpc.zookeeper.ZookeeperFactory"
destroy-method="close">
<property name="zkHosts"
value="192.168.36.54:2181,192.168.36.99:2181,192.168.36.189:2181" />
<property name="namespace" value="cn.slimsmart.thrift.rpc.demo" />
<property name="connectionTimeout" value="3000" />
<property name="sessionTimeout" value="3000" />
<property name="singleton" value="true" />
</bean>
<bean id="echoSerivce" class="cn.slimsmart.thrift.rpc.ThriftServiceClientProxyFactory" destroy-method="close">
<property name="maxActive" value="5" />
<property name="idleTime" value="1800000" />
<property name="serverAddressProvider">
<bean class="cn.slimsmart.thrift.rpc.zookeeper.ThriftServerAddressProviderZookeeper">
<property name="service" value="cn.slimsmart.thrift.rpc.demo.EchoSerivce" />
<property name="version" value="1.0.0" />
<property name="zkClient" ref="thriftZookeeper" />
</bean>
</property>
</bean>
</beans>

運行服務端后，我們可以看見zookeeper注冊了多個服務地址。

問題

安裝文檔： http://thrift.apache.org/docs/BuildingFromSource

遇到一個bison版本過低的問題： http://blog.csdn.net/qq910894904/article/details/41132779

參考資料

Thrift原理簡析（Java）－ http://shift-alt-ctrl.iteye.com/blog/1987416

Thrift－ http://www.cnblogs.com/mumuxinfei/tag/thrift/

Thrift與ZooKeeper－ http://blog.csdn.net/zhu_tianwei/article/details/44115667

Thrift專題－ http://blog.csdn.net/column/details/slimina-thrift.html

Thrift服務端與Spring－http://shift-alt-ctrl.iteye.com/blog/1990026

Thrift客戶端與Spring－http://shift-alt-ctrl.iteye.com/blog/1990030

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 thrift介紹 Thrift安裝介紹 Thrift架構介紹 Thrift總結(一)介紹 Apache Thrift安裝介紹 (ubuntu) RPC之Thrift 介紹及java實例全面介紹eBPF ZooKeeper全面介紹 Thrift介紹以及Java中使用Thrift實現RPC示例 Linux終端命令全面介紹