[转载] 基于zookeeper、连接池、Failover/LoadBalance等改造Thrift 服务化

标签：

转载自http://blog.csdn.net/zhu_tianwei/article/details/44115667

http://blog.csdn.net/column/details/slimina-thrift.html

对于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

[java] view plaincopyprint?

 

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包

[html] view plaincopyprint?

 

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

[java] view plaincopyprint?

 

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

2.服务端注册服务

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

ThriftServerIpResolve.java

[java] view plaincopyprint?

 

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

[java] view plaincopyprint?

 

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

[java] view plaincopyprint?

 

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

[java] view plaincopyprint?

 

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

[java] view plaincopyprint?

 

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

[java] view plaincopyprint?

 

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

[java] view plaincopyprint?

 

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

[java] view plaincopyprint?

 

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

[java] view plaincopyprint?

 

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

[html] view plaincopyprint?

 

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

[html] view plaincopyprint?

 

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注册了多个服务地址。

详细实例这里就不详述了，请参考实例代码：https://github.com/slimina/thrift-zookeeper-rpc

关于Thrift设计优化文档：

Thrift RPC服务框架日志的优化

[转载] 基于zookeeper、连接池、Failover/LoadBalance等改造Thrift 服务化

标签：

原文地址：http://www.cnblogs.com/scott19820130/p/4919174.html