Apache Thrift 初学小讲（四）【用Apache Commons Pool实现连接池】

Apache Thrift本身没有提供连接池，我们可以用Apache Commons Pool来实现一个，Apache commons-pool本质上是"对象池"，即通过一定的规则来维护对象集合的容器；commos-pool在很多场景中，用来实现连接池/任务worker池等，大家常用的dbcp数据库连接池，也是基于commons-pool实现的。TCP连接池里的是Socket对象，那么对于thirft连接池里的就是TTransport对象了。

参考了这篇文章：http://www.open-open.com/lib/view/open1415453575730.html，但上官网http://commons.apache.org/proper/commons-pool/download_pool.cgi下载jar时，发现官网上的例子的用法和上面那篇文章的例子的用法有点出入，才发现现在最新版是2.4.2，是重构过的，连包名pool都改成了pool2，不向下兼容的。官网poo2的例子：http://commons.apache.org/proper/commons-pool/examples.html。

据说poo2显著的提升了性能和可伸缩性，特别是在高并发加载的情况下。2.0 版本包含可靠的实例跟踪和池监控，要求 JDK 1.6 或者更新版本。poo2使用了泛型，不用再把object强制转换为你的对象类型了。

简单使用时主要涉及几个类：

GenericObjectPool：对象池类，可以获取和归还一个对象PooledObject等；

BasePooledObjectFactory：对象池GenericObjectPool里管理的对象的工厂类，池可以通过此工厂类创建和销毁一个对象PooledObject等；

GenericObjectPoolConfig：对象池GenericObjectPool的配置类，例如最大连接数，最大和最小空闲连接数等；

查看GenericObjectPool的构造方法，刚好两个参数分别是BasePooledObjectFactory和GenericObjectPoolConfig，而BasePooledObjectFactory是需要我们继承来实现相应的方法的，简单点实现两个抽象方法create和wrap即可，而wrap用默认实现new一个DefaultPooledObject就可以了。所以其实我们只需要实现create方法就行了。

有没有发现poo1是通过实现makeObject方法来创建池对象的，而poo2是create方法，同时poo2也有一个

makeObject方法，好像有点多余或者矛盾，我们查看poo2的BasePooledObjectFactory的源码发现其实poo2也是通过makeObject来创建一个池对象的，只不过里面依次调用了create和wrap，create了一个你的对象后，再把这个对象wrap成池管理类型的对象PooledObject。

源码：org.apache.commons.pool2.BasePooledObjectFactory-》makeObject：

package org.apache.commons.pool2;

public abstract class BasePooledObjectFactory<T> implements PooledObjectFactory<T> {
public abstract T create() throws Exception;
public abstract PooledObject<T> wrap(T obj);

@Override
public PooledObject<T> makeObject() throws Exception {
return wrap(create());
}

@Override
public void destroyObject(PooledObject<T> p)
throws Exception  {
}
@Override
public boolean validateObject(PooledObject<T> p) {
return true;
}
@Override
public void activateObject(PooledObject<T> p) throws Exception {
}
@Override
public void passivateObject(PooledObject<T> p)
throws Exception {
}
}

关门，放代码：

连接池工厂类ConnectionPoolFactory.java：

import org.apache.commons.pool2.BasePooledObjectFactory;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.apache.thrift.transport.TFastFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;

public class ConnectionPoolFactory {
private GenericObjectPool<TTransport> pool;

public ConnectionPoolFactory(GenericObjectPoolConfig config, String ip, int port) {
ConnectionFactory objFactory = new ConnectionFactory(ip, port, 3000);
pool = new GenericObjectPool<TTransport>(objFactory, config);
}

//从池里获取一个Transport对象
public TTransport getConnection() throws Exception {
return pool.borrowObject();
}

//把一个Transport对象归还到池里
public void releaseConnection(TTransport transport) {
pool.returnObject(transport);
}

/*
* 连接池管理的对象Transport的工厂类，
* GenericObjectPool会使用此类的create方法来
* 创建Transport对象并放进pool里进行管理等操作。
*/
class ConnectionFactory extends BasePooledObjectFactory<TTransport> {
private String ip;
private int port;
private int socketTimeout;

public ConnectionFactory(String ip, int port, int socketTimeout) {
this.ip = ip;
this.port = port;
this.socketTimeout = socketTimeout;
}

//创建TTransport类型对象方法
@Override
public TTransport create() throws Exception {
TSocket socket = new TSocket(ip, port);
socket.setTimeout(socketTimeout);
TTransport transport = socket;
transport = new TFastFramedTransport(transport);
if ( !transport.isOpen() ) {
transport.open();
}
return transport;
}

//把TTransport对象打包成池管理的对象PooledObject<TTransport>
@Override
public PooledObject<TTransport> wrap(TTransport transport) {
return new DefaultPooledObject<TTransport>(transport);
}

}
}

使用连接池的客户端类TestPoolClient.java：

import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TTransport;

import thrift.test.ThriftTest;
import thrift.test.User;

public class TestPoolClient {
public static void main(String [] args) throws Exception {
GenericObjectPoolConfig config = new GenericObjectPoolConfig();
//链接池中最大连接数,默认为8
config.setMaxTotal(2);
//当连接池资源耗尽时，调用者最大阻塞的时间，超时将跑出异常。单位：毫秒；默认为-1，表示永不超时
config.setMaxWaitMillis(3000);
ConnectionPoolFactory poolFactory = new ConnectionPoolFactory(config, "127.0.0.1", 9090);
//从连接池取得指定的通信方式
TTransport transport = poolFactory.getConnection();

//指定的通信协议
TProtocol tProtocol = new TCompactProtocol(transport);
//ThriftTest.Client是生成的客户端代码
ThriftTest.Client testClient = new ThriftTest.Client(tProtocol);

//getUser就是ThriftTest.thrift所定义的接口
User user = testClient.getUser(2);
System.out.println("名字："+ user.getName());

//归还transport到连接池
poolFactory.releaseConnection(transport);
}
}

运行TestServer.java，再运行TestPoolClient.java，结果：

（服务端打印）Starting the server on port 9090...

（客户端打印）名字：另外一个烟火

工程结构图：

附近src.rar是源码，感谢阅读！！！