commons-pool和commons-pool2是用来建立对象池的框架，提供了一些将对象池化必须要实现的接口和一些默认动作。对象池化之后可以通过pool的概念去管理其生命周期，例如对象的创建，使用，销毁等。例如我们通常使用的连接池，连接池可以有效管理连接的数量和状态，保证连接资源的情况而且避免并发场景下连接的频繁建立和释放。
我们这里来讲述如何使用commons-pool2来池化对象。我们以池化hadoop连接为例。
1、先解决依赖
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
<version>2.3</version>
</dependency>
2、如何使用连接池
我们是在spingboot框架中池化hadoop集群连接，先看一下池化之后的效果。
下面是我们池化之后的hadoop集群客户端。可以看到我们可以通过连接池的方式管理hadoo集群的链接。
1）配置连接池
最大连接数maxTotal
最大空闲连接数maxIdle
最小空闲连接数minIdle
获取连接的最大等待时间maxWait
可以看到传入这些配置的时候我们使用了一个config对象JHadoopPoolConfig，后面我们将说明这个config对象如何实现。
2）管理连接池
我们以三个函数说明了如何去连接池中申请连接，使用连接和释放链接资源。
申请资源pool.getResource()
释放资源pool.returnBrokenResource()和pool.returnResource()
这里要注意的是，一定要在catch和finally中成功释放资源，不然会导致could not get a Resource from the Pool的异常
package com.xiaoju.dqa.jazz.hadoop.client; import org.apache.hadoop.fs.FileStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import java.text.SimpleDateFormat;
import java.util.Date; public class JHadoopClient {
protected final Logger logger = LoggerFactory.getLogger(this.getClass()); private JHadoopPool jHadoopPool;
private String coreResource;
private String hdfsResource;
private int maxTotal;
private int maxIdle;
private int minIdle;
private int maxWaitMillis; public String getCoreResource() {
return coreResource;
}
public void setCoreResource(String coreResource) {
this.coreResource = coreResource;
}
public String getHdfsResource() {
return hdfsResource;
}
public void setHdfsResource(String hdfsResource) {
this.hdfsResource = hdfsResource;
}
public int getMaxTotal() {
return maxTotal;
}
public void setMaxTotal(int maxTotal) {
this.maxTotal = maxTotal;
}
public int getMaxIdle() {
return maxIdle;
}
public void setMaxIdle(int maxIdle) {
this.maxIdle = maxIdle;
}
public int getMaxWaitMillis() {
return maxWaitMillis;
}
public void setMaxWaitMillis(int maxWaitMillis) {
this.maxWaitMillis = maxWaitMillis;
}
public int getMinIdle() {
return minIdle;
}
public void setMinIdle(int minIdle) {
this.minIdle = minIdle;
} public void init() {
try {
JHadoopPoolConfig conf = new JHadoopPoolConfig();
conf.setMaxTotal(maxTotal);
conf.setMaxIdle(maxIdle);
conf.setMinIdle(minIdle);
conf.setMaxWaitMillis(maxWaitMillis);
jHadoopPool = new JHadoopPool(conf, coreResource, hdfsResource);
logger.info("[HDFS]初始化JHadoopClient成功");
} catch (Exception ex) {
logger.error("[HDFS]初始化JHadoopClient失败", ex);
}
} public void stop() {
try {
jHadoopPool.destroy();
} catch(Exception e) {
}
} public boolean exists(String path) throws Exception {
JHadoop jHadoop = null;
boolean broken = false;
try {
jHadoop = jHadoopPool.getResource();
return jHadoop.exists(path);
} catch (Exception e) {
broken = true;
jHadoopPool.returnBrokenResource(jHadoop);
logger.error("[HDFS]判断文件是否存在失败", e);
throw e;
} finally {
if (null != jHadoop && !broken) {
jHadoopPool.returnResource(jHadoop);
}
}
} public String getModificationTime(String path) throws Exception {
JHadoop jHadoop = null;
boolean broken = false;
try {
jHadoop = jHadoopPool.getResource();
FileStatus fileStatus = jHadoop.getFileStatus(path);
long modifyTimestamp = fileStatus.getModificationTime();
SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyyMMddHHmmss");
Date date = new Date(modifyTimestamp);
return simpleDateFormat.format(date);
} catch (Exception e) {
broken = true;
jHadoopPool.returnBrokenResource(jHadoop);
logger.error("[HDFS]获取最近修改时间失败", e);
throw e;
} finally {
if (null != jHadoop && !broken) {
jHadoopPool.returnResource(jHadoop);
}
}
} public long getPathSize(String path) throws Exception {
JHadoop jHadoop = null;
boolean broken = false;
try {
jHadoop = jHadoopPool.getResource();
return jHadoop.getContentSummary(path).getLength();
} catch (Exception e) {
broken = true;
jHadoopPool.returnBrokenResource(jHadoop);
logger.error("[HDFS]获取路径大小失败", e);
throw e;
} finally {
if (null != jHadoop && !broken) {
jHadoopPool.returnResource(jHadoop);
}
}
} }
3）注册成bean
通过配置文件传入链接池相应的配置。
package com.xiaoju.dqa.jazz.hadoop.configuration; import com.xiaoju.dqa.jazz.hadoop.client.JHadoopClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration; @Configuration
public class HadoopConfig { @Value("${hadoop.core.resource}")
private String coreResource;
@Value("${hadoop.hdfs.resource}")
private String hdfsResource;
@Value("${hadoop.pool.maxTotal}")
private int maxTotal;
@Value("${hadoop.pool.maxIdle}")
private int maxIdle;
@Value("${hadoop.pool.minIdle}")
private int minIdle;
@Value("${hadoop.pool.maxWaitMillis}")
private int maxWaitMillis; @Bean(initMethod = "init", destroyMethod = "stop")
public JHadoopClient jHadoopClient() {
JHadoopClient jHadoopClient = new JHadoopClient();
jHadoopClient.setMaxTotal(maxTotal);
jHadoopClient.setMaxIdle(maxIdle);
jHadoopClient.setMinIdle(minIdle);
jHadoopClient.setMaxWaitMillis(maxWaitMillis);
jHadoopClient.setCoreResource(coreResource);
jHadoopClient.setHdfsResource(hdfsResource);
return jHadoopClient;
}
}
4）config对象如何实现
我们这里要说明一下下面这些参数的含义：
1）setTestWhileConfig - 在空闲时检查有效性, 默认false
2）setMinEvictableIdleTimeMillis - 逐出连接的最小空闲时间
3）setTimeBetweenEvictionRunsMillis - 逐出扫描的时间间隔(毫秒) 如果为负数则不运行逐出线程，默认-1
4）每次逐出检查时 逐出的最大数目
package com.xiaoju.dqa.jazz.hadoop.client;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
public class JHadoopPoolConfig extends GenericObjectPoolConfig {
public JHadoopPoolConfig() {
this.setTestWhileIdle(true);
this.setMinEvictableIdleTimeMillis(60000L);
this.setTimeBetweenEvictionRunsMillis(30000L);
this.setNumTestsPerEvictionRun(-1);
}
}
3、连接池JHadoopPool
这个类继承了Pool<JHadoop>，用来初始化连接池对象。而JHadoop是Pool要管理的连接对象。
可以看到JHadoopPool在初始化的时候传入了一个JHadoopFactory的实例。这个实例将会以工厂模式来创建实际的JHadoop
JHadoopPool代码
package com.xiaoju.dqa.jazz.hadoop.client;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
public class JHadoopPool extends Pool<JHadoop> {
public JHadoopPool(GenericObjectPoolConfig poolConfig, String coreResource, String hdfsResource) {
super(poolConfig, new JHadoopFactory(coreResource, hdfsResource));
}
public JHadoopPool(GenericObjectPoolConfig poolConfig) {
super(poolConfig, new JHadoopFactory());
}
}
JHadoop代码
JHadoop实现了hadoop.fs中的方法调用。
我这里只给出了几个函数的简单封装，你可以根据具体的需要进行增加。
package com.xiaoju.dqa.jazz.hadoop.client; import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.ContentSummary;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import java.io.IOException; public class JHadoop {
protected final Logger logger = LoggerFactory.getLogger(this.getClass());
private FileSystem fs;
private String coreResource;
private String hdfsResource; public JHadoop(String coreResource, String hdfsResource) {
this.coreResource = coreResource;
this.hdfsResource = hdfsResource;
} public String getCoreResource() {
return coreResource;
}
public void setCoreResource(String coreResource) {
this.coreResource = coreResource;
}
public String getHdfsResource() {
return hdfsResource;
}
public void setHdfsResource(String hdfsResource) {
this.hdfsResource = hdfsResource;
} public void open() {
try {
Configuration conf = new Configuration();
conf.addResource(coreResource);
conf.addResource(hdfsResource);
fs = FileSystem.get(conf);
logger.info("[JHadoop]创建实例成功");
} catch (Exception e) {
logger.error("[JHadoop]创建实例失败", e);
}
} public void close() {
try {
if (null != fs) {
fs.close();
logger.info("[JHadoop]关闭实例成功");
}
} catch(Exception e) {
logger.error("[JHadoop]关闭实例失败", e);
}
} public boolean isConnected() throws IOException {
fs.exists(new Path("/forTest"));
return true;
} public boolean exists(String path) throws IOException {
Path hdfsPath = new Path(path);
return fs.exists(hdfsPath);
} public FileStatus getFileStatus(String path) throws IOException {
Path hdfsPath = new Path(path);
return fs.getFileStatus(hdfsPath);
} public ContentSummary getContentSummary(String path) throws IOException {
ContentSummary contentSummary = null;
Path hdfsPath = new Path(path);
if (fs.exists(hdfsPath)) {
contentSummary = fs.getContentSummary(hdfsPath);
}
return contentSummary;
} }
4、连接工厂类JHadoopFactory
JHadoopFactory这个类管理着连接对象的创建，销毁，验证等动作
package com.xiaoju.dqa.jazz.hadoop.client; import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.PooledObjectFactory;
import org.apache.commons.pool2.impl.DefaultPooledObject; public class JHadoopFactory implements PooledObjectFactory<JHadoop> {
private final String coreResource;
private final String hdfsResource; public JHadoopFactory() {
this.coreResource = "core-site.xml";
this.hdfsResource = "hdfs-site.xml";
}
public JHadoopFactory(String coreResource, String hdfsResource) {
this.coreResource = coreResource;
this.hdfsResource = hdfsResource;
} @Override
public PooledObject<JHadoop> makeObject() throws Exception {
JHadoop jHadoop = new JHadoop(coreResource, hdfsResource);
jHadoop.open();
return new DefaultPooledObject<JHadoop>(jHadoop);
} @Override
public void destroyObject(PooledObject<JHadoop> pooledJHadoop) throws Exception {
JHadoop jHadoop = pooledJHadoop.getObject();
jHadoop.close();
} @Override
public boolean validateObject(PooledObject<JHadoop> pooledJHadoop) {
JHadoop jHadoop = pooledJHadoop.getObject();
try {
return jHadoop.isConnected();
} catch (Exception e) {
return false;
}
} @Override
public void activateObject(PooledObject<JHadoop> pooledObject) throws Exception { } @Override
public void passivateObject(PooledObject<JHadoop> pooledObject) throws Exception { }
}