Java 如何快速实现一个连接池
作者:陈树义
什么是 ACP?
ACP 库提供了一整套用于实现对象池化的 API,以及若干种各具特色的对象池实现。目前最常用的版本是 2.0 版本,相对于 1.x 版本而言,并不是简单升级。2.0 版本是对象池实现的完全重写,显著的提升了性能和可伸缩性,并且包含可靠的实例跟踪和池监控。
Apache Commons Pool 的官网地址为:Pool – Overview,想翻找相关文档资料,到这里去是最权威、最全面的。
如何使用 ACP?
要使用 ACP 实现一个线程池,首先需要先引入 ACP 的依赖包,这里以 Maven 为例。
<dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-pool2</artifactId> <version>2.0</version> </dependency>
要使用 ACP 实现一个对象池,大致可以分为三个步骤:
- 创建对象工厂:告诉 ACP 如何创建你要的对象。
- 创建对象池:告诉 ACP 你想创建一个怎样的对象池。
- 使用对象池:ACP 告诉你如何使用你的对象。
创建对象工厂
对象工厂告诉 ACP,它应该如何去创建、激活、钝化、销毁你的对象。创建对象工厂非常简单,只需要实现 ACP 的 PooledObjectFactory 接口即可。PooledObjectFactory 接口的定义如下:
public interface PooledObjectFactory<T> { PooledObject<T> makeObject() throws Exception; void destroyObject(PooledObject<T> p) throws Exception; boolean validateObject(PooledObject<T> p); void activateObject(PooledObject<T> p) throws Exception; void passivateObject(PooledObject<T> p) throws Exception; }
但更多情况下,我们会继承 BasePooledObjectFactory 类来实现对象工厂。因为 BasePooledObjectFactory 类是 PooledObjectFactory 的基础实现类,使用它可以帮我们省了很多麻烦。通过继承这个抽象类,我们只需要实现两个方法:create() 和 wrap() 方法。
// 告诉 ACP 如何创建对象 public abstract T create() throws Exception; // 定义你要返回的对象 public abstract PooledObject<T> wrap(T obj);
create() 方法定义你的对象初始化过程,最后将初始化完成的对象返回。例如你想定义一个 SFTP 的连接,那么你首先需要定义一个 JSch 对象,之后设置账号密码,之后连接服务器,最后返回一个 ChannelSftp 对象。
public ChannelSftp create() { // SFTP 连接的创建过程 }
wrap() 方法定义你要返回的对象,对于一个 SFTP 的连接池来说,其实就是一个 ChannelSftp 对象。一般情况下可以使用类 DefaultPooledObject 替代,参考实现如下:
@Override public PooledObject<Foo> wrap(Foo foo) { return new DefaultPooledObject<Foo>(foo); }
创建对象池
创建好对象工厂之后,ACP 已经知道你需要的对象如何创建了。那么接下来,你需要根据你的实际需要,去创建一个对象池。在 ACP 中,我们通过 GenericObjectPool 以及 GenericObjectPoolConfig 来创建一个对象池。
// 声明一个对象池 private GenericObjectPool<ChannelSftp> sftpConnectPool; // 设置连接池配置 GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig(); poolConfig.setEvictionPolicyClassName("tech.shuyi.javacodechip.acp.SftpEvictionPolicy"); poolConfig.setBlockWhenExhausted(true); poolConfig.setJmxEnabled(false); poolConfig.setMaxWaitMillis(1000 * 10); poolConfig.setTimeBetweenEvictionRunsMillis(60 * 1000); poolConfig.setMinEvictableIdleTimeMillis(20 * 1000); poolConfig.setTestWhileIdle(true); poolConfig.setTestOnReturn(true); poolConfig.setTestOnBorrow(true); poolConfig.setMaxTotal(3); // 设置抛弃策略 AbandonedConfig abandonedConfig = new AbandonedConfig(); abandonedConfig.setRemoveAbandonedOnMaintenance(true); abandonedConfig.setRemoveAbandonedOnBorrow(true); this.sftpConnectPool = new GenericObjectPool<>(sftpConnectFactory, poolConfig, abandonedConfig);
在上面创建 SFTP 连接池的代码中,我们配置了一些线程池的参数以及设置了抛弃策略。抛弃策略是非常重要的,如果没有设置抛弃策略,那么会拿到失效的连接从而导致获取文件失败。抛弃策略是通过 poolConfig.setEvictionPolicyClassName 来设置的,我们这里设置的是 SftpEvictionPolicy 类,其代码内容如下:
@Slf4j @Component public class SftpEvictionPolicy implements EvictionPolicy<com.jcraft.jsch.ChannelSftp> { @Override public boolean evict(EvictionConfig config, PooledObject<com.jcraft.jsch.ChannelSftp> underTest, int idleCount) { try { // 连接失效时进行驱逐 if (!underTest.getObject().isConnected()) { log.warn("connect time out, evict the connection. time={}",System.currentTimeMillis() - underTest.getLastReturnTime()); return true; } }catch (Exception e){ return true; } return false; } }
看到这里,创建线程池的代码就结束了,SftpConnectPool 文件的全部内容如下:
@Slf4j public class SftpConnectPool { private GenericObjectPool<ChannelSftp> sftpConnectPool; public SftpConnectPool(SftpConnectFactory sftpConnectFactory) { // 设置连接池配置 GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig(); poolConfig.setEvictionPolicyClassName("tech.shuyi.javacodechip.acp.SftpEvictionPolicy"); poolConfig.setBlockWhenExhausted(true); poolConfig.setJmxEnabled(false); poolConfig.setMaxWaitMillis(1000 * 10); poolConfig.setTimeBetweenEvictionRunsMillis(60 * 1000); poolConfig.setMinEvictableIdleTimeMillis(20 * 1000); poolConfig.setTestWhileIdle(true); poolConfig.setTestOnReturn(true); poolConfig.setTestOnBorrow(true); poolConfig.setMaxTotal(3); // 设置抛弃策略 AbandonedConfig abandonedConfig = new AbandonedConfig(); abandonedConfig.setRemoveAbandonedOnMaintenance(true); abandonedConfig.setRemoveAbandonedOnBorrow(true); this.sftpConnectPool = new GenericObjectPool<>(sftpConnectFactory, poolConfig, abandonedConfig); } public ChannelSftp borrowObject() { try { return sftpConnectPool.borrowObject(); } catch (Exception e) { log.error("borrowObject error", e); return null; } } public void returnObject(ChannelSftp channelSftp) { if (channelSftp!=null) { sftpConnectPool.returnObject(channelSftp); } } }
为了方便使用,我还增加了 borrowObject 和 returnObject 方法,但这两个并不是必须的。在这两个方法中,我们分别调用了 GenericObjectPool 类的 borrowObject 方法和 returnObject 方法。这正是 ACP 提供的、使用线程池对象的方法,先借一个对象,之后归还对象。
注:其实在这一步,已经包含了对象池的使用了。但实际使用的时候,我们经常是将对象池的声明与使用放在同一个类中,因此为了讲解方便,这里没有分开。因此下文的使用对象池,本质上是对对象池做进一步封装。
使用对象池
到这里我们的 SFTP 对象池就已经创建完毕了,是不是非常简单呢!但在实际的工作中,我们通常会在这基础上,做一些封装。对于我们这次的 SFTP 连接池来说,我们会对外直接提供下载文件的服务,将 SFTP 对象池进一步封装起来,不需要关心怎么获取文件。
public class SftpFileHelper { @Autowired private SftpConnectPool sftpConnectPool; public void download(String dir, String file, String saveUrl)throws IOException { ChannelSftp sftp = sftpConnectPool.borrowObject(); log.info("begin to download file, dir={}, file={}, saveUrl={}", dir, file, saveUrl); try { if (!StringUtils.isEmpty(dir)) { sftp.cd(dir); } File downloadFile = new File(saveUrl); sftp.get(file, new FileOutputStream(downloadFile)); }catch (Exception e){ log.warn("下载文件失败", e); }finally { sftpConnectPool.returnObject(sftp); } log.info("file:{} is download successful", file); } }
最后我们写一个测试用例来试一试,是否能正常下载文件。
@RunWith(SpringRunner.class) @SpringBootTest @Slf4j public class SftpFileHelperTest { @Autowired private SftpFileHelper sftpFileHelper; @Test public void testDownloadFtpFile() throws Exception { sftpFileHelper.download("dir", "fileName", "fileName"); } }
总结
本文针对 Apache Commons Pool 库最常用的对象池功能做了演示。看完这篇文章,我们知道创建一个线程池需要三个步骤,分别是:
- 创建对象工厂:告诉 ACP 如何创建你要的对象。
- 创建对象池:告诉 ACP 你想创建一个怎样的对象池、设置驱逐策略。
- 使用对象池:ACP 告诉你如何使用你的对象。
本文相关代码存放在博主 Github 项目:java-code-chip 中,可以点击地址获取:java-code-chip/src/main/java/tech/shuyi/javacodechip/acp at master · chenyurong/java-code-chip
ACP 库能够让读者朋友们快速地创建一个对象池,更加专注于业务内容。但事实上,ACP 提供的内容远不止如此,它还有更多更高级的功能。
例如当我们连接的 SFTP 服务器有多个时,我们需要通过不同地址来获得不同的连接对象。此时最笨的办法是每个不同的地址,都复制多一份代码,然后通过不同类的不同方法来实现。但这样的情况工作量相当可观,并且也会有很多重复代码。这种时候就可以使用BaseKeyedPooledObjectFactory 来替代 BasePooledObjectFactory,从而实现通过 key 来实现不同地址的连接对象管理。
更多关于 ACP 的内容,感兴趣的同学可以自行探索,这里就不深入讲解了。
另一种实现方式:
泛型接口ConnectionPool.java
public interface ConnectionPool<T> { /** * 初始化池资源 * @param maxActive 池中最大活动连接数 * @param maxWait 最大等待时间 */ void init(Integer maxActive, Long maxWait); /** * 从池中获取资源 * @return 连接资源 */ T getResource() throws Exception; /** * 释放连接 * @param connection 正在使用的连接 */ void release(T connection) throws Exception; /** * 释放连接池资源 */ void close(); }
以zookeeper为例,实现zookeeper连接池,ZookeeperConnectionPool.java
public class ZookeeperConnectionPool implements ConnectionPool<ZooKeeper> { //最大活动连接数 private Integer maxActive; //最大等待时间 private Long maxWait; //空闲队列 private LinkedBlockingQueue<ZooKeeper> idle = new LinkedBlockingQueue<>(); //繁忙队列 private LinkedBlockingQueue<ZooKeeper> busy = new LinkedBlockingQueue<>(); //连接池活动连接数 private AtomicInteger activeSize = new AtomicInteger(0); //连接池关闭标记 private AtomicBoolean isClosed = new AtomicBoolean(false); //总共获取的连接记数 private AtomicInteger createCount = new AtomicInteger(0); //等待zookeeper客户端创建完成的计数器 private static ThreadLocal<CountDownLatch> latchThreadLocal = ThreadLocal.withInitial(() -> new CountDownLatch(1)); public ZookeeperConnectionPool(Integer maxActive, Long maxWait) { this.init(maxActive, maxWait); } @Override public void init(Integer maxActive, Long maxWait) { this.maxActive = maxActive; this.maxWait = maxWait; } @Override public ZooKeeper getResource() throws Exception { ZooKeeper zooKeeper; Long nowTime = System.currentTimeMillis(); final CountDownLatch countDownLatch = latchThreadLocal.get(); //空闲队列idle是否有连接 if ((zooKeeper = idle.poll()) == null) { //判断池中连接数是否小于maxActive if (activeSize.get() < maxActive) { //先增加池中连接数后判断是否小于等于maxActive if (activeSize.incrementAndGet() <= maxActive) { //创建zookeeper连接 zooKeeper = new ZooKeeper("localhost", 5000, (watch) -> { if (watch.getState() == Watcher.Event.KeeperState.SyncConnected) { countDownLatch.countDown(); } }); countDownLatch.await(); System.out.println("Thread:" + Thread.currentThread().getId() + "获取连接:" + createCount.incrementAndGet() + "条"); busy.offer(zooKeeper); return zooKeeper; } else { //如增加后发现大于maxActive则减去增加的 activeSize.decrementAndGet(); } } //若活动线程已满则等待busy队列释放连接 try { System.out.println("Thread:" + Thread.currentThread().getId() + "等待获取空闲资源"); Long waitTime = maxWait - (System.currentTimeMillis() - nowTime); zooKeeper = idle.poll(waitTime, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { throw new Exception("等待异常"); } //判断是否超时 if (zooKeeper != null) { System.out.println("Thread:" + Thread.currentThread().getId() + "获取连接:" + createCount.incrementAndGet() + "条"); busy.offer(zooKeeper); return zooKeeper; } else { System.out.println("Thread:" + Thread.currentThread().getId() + "获取连接超时,请重试!"); throw new Exception("Thread:" + Thread.currentThread().getId() + "获取连接超时,请重试!"); } } //空闲队列有连接,直接返回 busy.offer(zooKeeper); return zooKeeper; } @Override public void release(ZooKeeper connection) throws Exception { if (connection == null) { System.out.println("connection 为空"); return; } if (busy.remove(connection)){ idle.offer(connection); } else { activeSize.decrementAndGet(); throw new Exception("释放失败"); } } @Override public void close() { if (isClosed.compareAndSet(false, true)) { idle.forEach((zooKeeper) -> { try { zooKeeper.close(); } catch (InterruptedException e) { e.printStackTrace(); } }); busy.forEach((zooKeeper) -> { try { zooKeeper.close(); } catch (InterruptedException e) { e.printStackTrace(); } }); } } }
测试用例
这里创建20个线程并发测试连接池,Test.java
public class Test { public static void main(String[] args) throws Exception { int threadCount = 20; Integer maxActive = 10; Long maxWait = 10000L; ZookeeperConnectionPool pool = new ZookeeperConnectionPool(maxActive, maxWait); CountDownLatch countDownLatch = new CountDownLatch(20); for (int i = 0; i < threadCount; i++) { new Thread(() -> { countDownLatch.countDown(); try { countDownLatch.await(); ZooKeeper zooKeeper = pool.getResource(); Thread.sleep(2000); pool.release(zooKeeper); } catch (InterruptedException e) { e.printStackTrace(); } catch (Exception e) { e.printStackTrace(); } }).start(); } while (true){ } } }
以上就是Java 如何快速实现一个连接池的详细内容,更多关于Java 连接池的资料请关注脚本之家其它相关文章!