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&lt;T&gt; {

    /**
     * 初始化池资源
     * @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&lt;ZooKeeper&gt; {
    //最大活动连接数
    private Integer maxActive;
    //最大等待时间
    private Long maxWait;
    //空闲队列
    private LinkedBlockingQueue&lt;ZooKeeper&gt; idle = new LinkedBlockingQueue&lt;&gt;();
    //繁忙队列
    private LinkedBlockingQueue&lt;ZooKeeper&gt; busy = new LinkedBlockingQueue&lt;&gt;();
    //连接池活动连接数
    private AtomicInteger activeSize = new AtomicInteger(0);
    //连接池关闭标记
    private AtomicBoolean isClosed = new AtomicBoolean(false);
    //总共获取的连接记数
    private AtomicInteger createCount = new AtomicInteger(0);
    //等待zookeeper客户端创建完成的计数器
    private static ThreadLocal&lt;CountDownLatch&gt; latchThreadLocal = ThreadLocal.withInitial(() -&gt; 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() &lt; maxActive) {
                //先增加池中连接数后判断是否小于等于maxActive
                if (activeSize.incrementAndGet() &lt;= maxActive) {
                    //创建zookeeper连接
                    zooKeeper = new ZooKeeper("localhost", 5000, (watch) -&gt; {
                        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) -&gt; {
                try {
                    zooKeeper.close();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
            busy.forEach((zooKeeper) -&gt; {
                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 &lt; threadCount; i++) {
            new Thread(() -&gt; {
                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 连接池的资料请关注我们其它相关文章!

(0)

相关推荐

  • 如何解决线程太多导致java socket连接池出现的问题

    这篇文章主要介绍了如何解决线程太多导致socket连接池出现的问题,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 线程太多导致socket连接池爆满,进程启动不了 问题: 某部机上跟其它机器的连接有问题,ping可以通,telnet端口不通,可以其它机器可以连接到该机器上的进程. java应用启动不起来,产生以下错误. java.net.SocketException: No buffer space available (maximum co

  • Java FTPClient连接池的实现

    最近在写一个FTP上传工具,用到了Apache的FTPClient,为了提高上传效率,我采用了多线程的方式,但是每个线程频繁的创建和销毁FTPClient对象势必会造成不必要的开销,因此,此处最好使用一个FTPClient连接池.仔细翻了一下Apache的api,发现它并没有一个FTPClientPool的实现,所以,不得不自己写一个FTPClientPool.下面就大体介绍一下开发连接池的整个过程,供大家参考. 关于对象池 有些对象的创建开销是比较大的,比如数据库连接等.为了减少频繁创建.销毁

  • Java数据库连接池技术的入门教程

    池(Pool)技术在一定程度上可以明显优化服务器应用程序的性能,提高程序执行效率和降低系统资源开销.数据库连接池负责分配.管理和释放数据库连接,它允许应用程序重复使用一个现有的数据库连接,而不是再重新建立一个: 数据库连接池技术 基本概念: 数据库连接池:其实就是一个容器(集合),存放数据库连接的容器. 当系统初始化好后,容器被创建,容器中会申请一些连接对象,当用户来访问数据库时,从容器中获取连接对象,用户访问完之后,会将连接对象归还给容器. 使用池化技术的好处: 1. 节约资源 2. 用户访问

  • 浅谈常用Java数据库连接池(小结)

    概述 在这里所谓的数据库连接是指通过网络协议与数据库服务之间建立的TCP连接.通常,与数据库服务进行通信的网络协议无需由应用程序本身实现,原因有三: 1.实现复杂度大,需要充分理解和掌握相应的通信协议. 2.代码难以复用,每个应用程序都需要独立实现一套对应的网络协议(不同公司之间,同一公司的不同技术栈之间难以复用实现相同协议的代码) 3.性能难以保证,不同的网络协议实现可能存在巨大的性能差距 正因为如此,所以现实的实现方式是: 首先,定义网络协议标准,这样只要支持这个标准协议的数据库就可以使用相

  • Java数据库连接池连接Oracle过程详解

    第一步:导入tomcat\lib 下的一个tomcat-dbcp.jar包 第二步:在web\META-INF下新建一个context.xml文件,文件内容如下: <?xml version="1.0" encoding="UTF-8"?> <Context> <Resource name="jdbc/lwl" auth="Container" type="javax.sql.DataS

  • Java使用MySQL实现连接池代码实例

    官方:数据库连接池(Connection pooling)是程序启动时建立足够的数据库连接,并将这些连接组成一个连接池,由程序动态地对连接池中的连接进行申请,使用,释放. 理解:创建数据库连接池是一个很耗时的操作,也容易对数据库造成安全隐患.所以,在程序初始化的时候,集中创建多个数据库连接池,并把他们集中管理,供程序使用,可以保证较快的数据库读写速度,还更加的安全可靠. 手动配置连接池: /** * 手动设置连接池 */ public void demo1(){ // 获得连接: Connect

  • Java dbcp连接池基本使用方法详解

    1.依赖api的使用 导入jar包 <!-- https://mvnrepository.com/artifact/org.apache.commons/commons-dbcp2 --> <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-dbcp2</artifactId> <version>2.7.0</version&g

  • Java 模拟数据库连接池的实现代码

    前面学习过等待 - 通知机制,现在我们在其基础上添加一个超时机制,模拟从连接池中获取.使用和释放连接的过程.客户端获取连接的过程被设定为等待超时模式,即如果在 1000 毫秒内无法获取到可用连接,将会返回给客户端一个 null.设定连接池的大小为 10 个,然后通过调节客户端的线程数来模拟无法获取连接的场景 由于 java.sql.Connection 只是一个接口,最终实现是由数据库驱动提供方来实现,考虑到本例只是演示,我们通过动态代理构造一个 Connection,该 Connection

  • Java中Druid连接池连接超时获取不到连接的解决

    错误内容: com.alibaba.druid.pool.GetConnectionTimeoutException: wait millis 30000, active 600, maxActive 600, creating 0 detail: Service Error:Cannot find a proper coonection from STDB 错误日志截图: 解决过程: 1.添加了三个参数 作用是如果超过3分钟,连接未释放,那么关闭连接,并报错. 2.进行请求,并查看日志 确认获

  • Java数据库连接池c3p0过程解析

    首先下载c3p0的jar包 https://mvnrepository.com/search?q=c3p0 导入jar包到eclipse 将c3p0的jar包放到eclipse中,单击右键,选择build path ,在选择 add jars 创建连接池对象 ComboPooledDataSource cpd= new ComboPooledDataSource(); 设置连接参数 方法一 使用静态代码块来初始化参数(不推荐)如果数据库发生改变,源文件需要重新编写编译,项目需要重新部署 stat

随机推荐