java连接池详解与自定义es连接池

目录1 版本选择2 导入依赖3 使用commons-pool构造连接池3.1 pom.xml3.2 对象池类GenericObjectPool普通对象池GenericKeyedObjectPool带Key的对象池3.3 实现PoolableObjectFactory接口类3.4 继承BasePooledObjectFactory抽象类3.5 GenericObjectPool.Config()参数

进击的豌豆哥

3501人浏览 · 2021-11-10 16:45:16

进击的豌豆哥 · 2021-11-10 16:45:16 发布

我们在服务器开发的过程中，往往会有一些对象，它的创建和初始化需要的时间比较长，比如数据库连接，网络IO，大数据对象等。在大量使用这些对象时，如果不采用一些技术优化，就会造成一些不可忽略的性能影响。一种办法就是使用对象池，每次创建的对象并不实际销毁，而是缓存在对象池中，下次使用的时候，不用再重新创建，直接从对象池的缓存中取即可。为了避免重新造轮子，我们可以使用优秀的开源对象池化组件apache-commons-pool2，它对对象池化操作进行了很好的封装，我们只需要根据自己的业务需求重写或实现部分接口即可，使用它可以快速的创建一个方便，简单，强大对象连接池管理类。

1 版本选择

可以看出有两种版本的依赖可以导入，分别是commons-pool2和commons-pool
commons-pool2是commons-pool的升级版。
数据库连接池dbcp的版本一般与pool的版本对应

2 依赖选择

老规矩，选择引用次数最多的依赖进行导入

apache连接池:
commons-dbcp 对应的连接池是 commons-pool，commons-pool版本最高到1.6(2010年)，再往上就在pool2上发展从2.0开始。
相应的数据库连接池dbcp，对应pool版本，dbcp最高版本1.4，升级到pool2后，也升级为dbcp2，版本从2.0开始。
配置时，版本不要对错，否则系统会出错。

commons-pool2


对象池化组件
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-pool2</artifactId>
    <version>2.9.0</version>
</dependency>

数据库连接池
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-dbcp2</artifactId>
    <version>2.7.0</version>
</dependency>

es依赖
<!-- https://mvnrepository.com/artifact/org.elasticsearch/elasticsearch -->
<dependency>
   <groupId>org.elasticsearch</groupId>
   <artifactId>elasticsearch</artifactId>
   <version>6.5.4</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.elasticsearch.client/elasticsearch-rest-high-level-client -->
<dependency>
   <groupId>org.elasticsearch.client</groupId>
   <artifactId>elasticsearch-rest-high-level-client</artifactId>
   <version>6.5.4</version>
</dependency>

commons-pool


对象池化组件
 <dependency>
     <groupId>commons-pool</groupId>
     <artifactId>commons-pool</artifactId>
     <version>1.6</version>
 </dependency>
 
数据库连接池
 <dependency>
     <groupId>commons-dbcp</groupId>
     <artifactId>commons-dbcp</artifactId>
     <version>1.4</version>
 </dependency>
 
es依赖
<!-- https://mvnrepository.com/artifact/org.elasticsearch/elasticsearch -->
<dependency>
   <groupId>org.elasticsearch</groupId>
   <artifactId>elasticsearch</artifactId>
   <version>6.5.4</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.elasticsearch.client/elasticsearch-rest-high-level-client -->
<dependency>
   <groupId>org.elasticsearch.client</groupId>
   <artifactId>elasticsearch-rest-high-level-client</artifactId>
   <version>6.5.4</version>
</dependency>

3 使用commons-pool构造连接池

3.1 pom.xml

 <dependency>
     <groupId>commons-pool</groupId>
     <artifactId>commons-pool</artifactId>
     <version>1.6</version>
 </dependency>

3.2 对象池类

GenericObjectPool普通对象池

这个是对象池实现的核心类，它实现了对对象池的管理，是一个基本的对象池实现，一般情况下，我们可以直接使用。在使用这个类的时候，我们需要传入两个重要的参数：GenericObjectPoolConfig类和PooledObjectFactory接口的实现，一会我们再详细说这两个。
在GenericObjectPool中，有两个我们会用到的方法：

// 对象使用完之后，归还到对象池
@Override
public void returnObject(final T obj) {
    .....
}
// 从对象池中获取一个对象
@Override
public T borrowObject() throws Exception {
    return borrowObject(getMaxWaitMillis());
}

其它还有一些方法，比如关闭对象池，销毁对象池，获取对象池中空闲的对象个数等，可以自行查看API。

GenericKeyedObjectPool带Key的对象池

这种对象池和前面的GenericObjectPool对象池操作是一样的，不同的是对应的每个方法带一个key参数。你可以把这个GenericKeyedObjectPool的对象池看作是一个map的GenericObjectPool，每个key对应一个GenericObjectPool。它用于区别不同类型的对象。比如数据库连接，有可能会连接到不同地址的数据库上面。就可以用这个区分。

3.3 实现PoolableObjectFactory接口类

实现PoolableObjectFactory接口，这个接口是我们要实现的，它对要实现对象池化的对象做了一些管理。这个工厂接口就是为了让我们根据自己的业务创建和管理要对象池化的对象。当然也可以继承默认的抽象类：BasePooledObjectFactory .

PoolableObjectFactory接口方法解读：

PooledObject<T> makeObject() throws Exception;

这个方法是用来创建一个对象，当在GenericObjectPool类中调用borrowObject方法时，如果当前对象池中没有空闲的对象，GenericObjectPool会调用这个方法，创建一个对象，并把这个对象封装到PooledObject类中，并交给对象池管理。

void destroyObject(PooledObject<T> p) throws Exception;

销毁对象，当对象池检测到某个对象的空闲时间(idle)超时，或使用完对象归还到对象池之前被检测到对象已经无效时，就会调用这个方法销毁对象。对象的销毁一般和业务相关，但必须明确的是，当调用这个方法之后，对象的生命周期必须结果。如果是对象是线程，线程必须已结束，如果是socket，socket必须已close，如果是文件操作，文件数据必须已flush，且文件正常关闭。

boolean validateObject(PooledObject<T> p);

检测一个对象是否有效。在对象池中的对象必须是有效的，这个有效的概念是，从对象池中拿出的对象是可用的。比如，如果是socket,那么必须保证socket是连接可用的。在从对象池获取对象或归还对象到对象池时，会调用这个方法，判断对象是否有效，如果无效就会销毁。

void activateObject(PooledObject<T> p) throws Exception;

激活一个对象或者说启动对象的某些操作。比如，如果对象是socket，如果socket没有连接，或意外断开了，可以在这里启动socket的连接。它会在检测空闲对象的时候，如果设置了测试空闲对象是否可以用，就会调用这个方法，在borrowObject的时候也会调用。另外，如果对象是一个包含参数的对象，可以在这里进行初始化。让使用者感觉这是一个新创建的对象一样。

void passivateObject(PooledObject<T> p) throws Exception;

钝化一个对象。在向对象池归还一个对象是会调用这个方法。这里可以对对象做一些清理操作。比如清理掉过期的数据，下次获得对象时，不受旧数据的影响。

一般来说activateObject和passivateObject是成对出现的。前者是在对象从对象池取出时做一些操作，后者是在对象归还到对象池做一些操作，可以根据自己的业务需要进行取舍。

3.4 继承BasePooledObjectFactory抽象类

这个抽象类是上面PooledObjectFactory 接口的空现，并且透出了两个抽象方法必须实现。

public abstract T create() throws Exception;

创建一个对象实例，可以被wrap成一个PooledObject，这个方法必须支持并发和多线程。

public abstract PooledObject<T> wrap(T obj);

把一个对象包装为一个PooledObject，此方法只在调用borrowObject方法的时候，且返回一个全新对象的时候执行，此方法处理create()方法的返回值。常见的处理方式是new DefaultPooledObject<>(obj)。可以在包装前进行其他逻辑的处理

3.5 GenericObjectPool.Config()参数配置类

这个类允许使用者对对象池的一些参数进行调整，根据需要定制对象池。下面说逐一说一下每个参数的含义。

基础配置

maxTotal：对象池中最大对象数量。默认值是8。
maxIdle：对象池中最大的空闲对象个数。默认值是8，那么当连接池中全部空闲时，会关闭maxTotal-maxIdle条的空闲连接，存留maxIdle条连接。

对象池中最大空闲对象数量是指，在对象池中维护的空闲对象数量不能超过这个值。当对象池中的空闲对象数量达到最大值时，池会根据策略来销毁多余的对象，以防止对象池中的对象数量过多，浪费内存资源。
这个参数的作用是控制对象池中空闲对象的数量，避免对象池中的对象数量过多，浪费内存资源。例如，如果一个Web应用程序使用了对象池来维护数据库连接对象，可以设置最大空闲对象数量，避免在低负载情况下对象池中的连接对象数量过多，从而浪费内存资源。
需要注意的是，如果应用程序在高负载情况下需要更多的对象，而对象池中的空闲对象数量已经达到最大值，那么应用程序可能会阻塞或超时。因此，需要根据应用程序的负载情况和硬件资源的限制来设置最大空闲对象数量。
minIdle：对象池中最小的空闲对象个数。也就是说，如果这个值不为0（必须要小于maxidle），并且当连接池开启周期检测时，检测到连接大量失效，那么会销毁失效的连接，如果存活的连接数小于了minidle，那么则进行创建，保证连接池中至少存在着minidle条连接。默认值是0。
testOnCreate：在创建对象时检测对象是否有效，true是，默认值是false, 配置true会降低性能。
testOnBorrow: 向调用者输出“链接”资源时，是否检测是有有效, 如果无效则从连接池中移除，并尝试获取继续获取，配置true会降低性能，如果无效则从连接池中移除，并尝试获取继续获取。默认为false。建议保持默认值.如果为true，需要配合以下代码完成有效性验证，在PooledObjectFactory实现类中。

validateObject会在执行GenericObjectPool的borrowObject方法前执行
```
	@SneakyThrows
	@Override
	public boolean validateObject(PooledObject<RestHighLevelClient> p) {
	    System.out.println("检测是否有效");
	    RestHighLevelClient client = p.getObject();
	    MainResponse info = client.info(RequestOptions.DEFAULT);
	    return StringUtils.isNotBlank(info.getClusterUuid());
	}
```
blockWhenExhausted：当对象池没有空闲对象时，新的获取对象的请求是否阻塞。true阻塞。默认值是true;
当对象池没有空闲对象时，新的获取对象的请求是否阻塞（true 阻塞，maxWaitMillis 才生效； false 连接池没有资源立马抛异常）
maxWaitMillis：需要配合blockWhenExhausted使用，当没有空闲连接时，获取一个对象的最大等待时间。如果这个值小于0，则永不超时，一直等待，直到有空闲对象到来。如果大于0，则等待maxWaitMillis长时间，如果没有空闲对象，将抛出NoSuchElementException异常。默认值是-1；可以根据需要自己调整，单位是毫秒。
whenExhaustedAction: 当“连接池”中active数量达到阀值时，即“链接”资源耗尽时，连接池需要采取的手段, 默认为1：
-> 0 : 抛出异常，
-> 1 : 阻塞，直到有可用链接资源
-> 2 : 强制创建新的链接资源
testOnReturn: 向连接池“归还”链接时，是否检测“链接”对象的有效性, 配置true会降低性能。默认为false。建议保持默认值.

其他配置

lifo：对象池存储空闲对象是使用的LinkedBlockingDeque，它本质上是一个支持后进先出(LIFO)与先进先出(FIFO)两种行为的双向的队列，common-pool2中的LinkedBlockingDeque不是Java原生的队列，而有common-pool2重新写的一个双向队列。如果为true，表示使用FIFO后进先出获取对象。默认值是true.建议使用默认值。
fairness：当从池中获取资源或者将资源还回池中时,是否使用java.util.concurrent.locks.ReentrantLock.ReentrantLock 的公平锁机制。common-pool2实现的LinkedBlockingDeque双向阻塞队列使用的是Lock锁。这个参数就是表示在实例化一个LinkedBlockingDeque时，是否使用lock的公平锁。默认值是false，建议使用默认值。
numTestsPerEvictionRun：检测空闲对象线程每次检测的空闲对象的数量。默认值是3；如果这个值小于0,则每次检测的空闲对象数量等于当前空闲对象数量除以这个值的绝对值，并对结果向上取整。
timeBetweenEvictionRunsMillis：空闲对象检测线程的执行周期，即多长时候执行一次空闲对象检测。单位是毫秒数。如果小于等于0，则不执行检测线程。默认值是-1;

注意
上面配置已经足够，如下配置未经过测试，不要用于生产环境中

若想保存maxIdle个连接数，则需要进行如下配置

在测试中，下面两个参数的作用相同，并且没有起到作用，实际上在连接池空闲时就自动保存maxidle个连接，后续有待考察

minEvictableIdleTimeMillis：连接的最小空闲时间，达到此值后该空闲连接可能会被移除（还需看是否已达最大空闲连接数）；
默认值 DEFAULT_MIN_EVICTABLE_IDLE_TIME_MILLIS = 1000L * 60L * 30L
softMinEvictableIdleTimeMillis：连接空闲的最小时间，达到此值后空闲链接将会被移除，且保留minIdle个空闲连接数；
默认值 DEFAULT_SOFT_MIN_EVICTABLE_IDLE_TIME_MILLIS = -1

若想进行周期连接有效性检测（周期性检测连接池中对象的状态，对于失效的连接则进行销毁和重新创建，最终存留minidle条连接，则需要进行如下配置
需要配合以下代码完成有效性验证，在PooledObjectFactory实现类中。

@SneakyThrows
@Override
public boolean validateObject(PooledObject<RestHighLevelClient> p) {
    System.out.println("检测是否有效");
    RestHighLevelClient client = p.getObject();
    MainResponse info = client.info(RequestOptions.DEFAULT);
    return StringUtils.isNotBlank(info.getClusterUuid());
}

testWhileIdle: 在检测空闲对象线程检测到对象不需要移除时，是否检测对象的有效性，与timeBetweenEvictionRunsMillis配合使用。配置为true时，不影响性能，并且保证安全性，但是会在相应时间内，取出连接池内numTestsPerEvictionRun个连接对象检测他们的是否存活，这个值最大为minIdle，即最小连接数。
timeBetweenEvictionRunsMillis：空闲连接检测的周期（单位毫秒）；与testWhileIdle配合使用，如果为负值，表示不运行检测线程；
默认值 DEFAULT_TIME_BETWEEN_EVICTION_RUNS_MILLIS = -1L
numTestsPerEvictionRun
检测空闲对象线程每次运行时检测的空闲对象的数量；
如果 numTestsPerEvictionRun>=0, 则取numTestsPerEvictionRun 和池内的连接数的较小值作为每次检测的连接数；
如果 numTestsPerEvictionRun<0，则每次检查的连接数是检查时池内连接的总数除以这个值的绝对值再向上取整的结果；
默认值 DEFAULT_NUM_TESTS_PER_EVICTION_RUN = 3
evictorShutdownTimeoutMillis：关闭驱逐线程的超时时间；
默认值 DEFAULT_EVICTOR_SHUTDOWN_TIMEOUT_MILLIS = 10L * 1000L

以上就是common-pool对象池的配置参数，使用的时候可以根据自己的需要进行调整。

配置的注意事项

private RestClientBuilder getRestClientBuilder() {
    String[] hosts = this.hosts.split(",");
    HttpHost[] httpHosts = new HttpHost[hosts.length];
    for (int i = 0; i < hosts.length; i++) {
        httpHosts[i] = new HttpHost(hosts[i], port, "http");
    }

    final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
    credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));

    return RestClient.builder(httpHosts).setRequestConfigCallback(requestConfigBuilder -> {
        requestConfigBuilder.setConnectTimeout(-1); // 连接超时（默认为1秒）
        // * 如果线程池内对象已经无效了 则用这个时间判断 多久没响应 就执行销毁方法 然后取新的 直到池中没有了 然后创建
        // 应该配合maxidle大小来计算
        requestConfigBuilder.setSocketTimeout(3000);//套接字超时（默认为30秒）//更改客户端的超时限制默认30秒现在改为100*1000分钟
        requestConfigBuilder.setConnectionRequestTimeout(5000);
        return requestConfigBuilder;
    }).setHttpClientConfigCallback(httpClientBuilder -> {
        httpClientBuilder.disableAuthCaching();
        return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
    });
}

假如我现在连接池中有8个存活连接，过了半个小时，实际上这8个连接已经不能用了，但是还在连接池中，这时候我们来连接池取连接的时候，就会卡在activateObject方法，直到SocketTimeout的时间（默认20s）没有响应，才会销毁这个连接，执行destroyObject方法，然后取下一个连接，循环往复，直到连接池出现有效的连接或者连接池没有连接了，创建新的连接。这时候就会等很长时间，8x20，也就是120秒，此时我们可以调小连接池的maxidle参数和setSocketTimeout时间，比如4个，3000毫秒，这时候，在理论上，即使连接池内连接全部失效，我们也只会等待4x3=12秒

3.6 elasticsearch连接池代码实现

ElasticSearchPoolUtil.java


import org.apache.commons.pool.impl.GenericObjectPool;
import org.elasticsearch.client.RestHighLevelClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * ElasticSearch 连接池工具类
 */
public class ElasticSearchPoolUtil {

    private static Logger logger = LoggerFactory.getLogger(ElasticSearchPoolUtil.class);

    // 对象池配置类，不写也可以，采用默认配置
    private static GenericObjectPool.Config poolConfig = new GenericObjectPool.Config();

    // 采用默认配置maxTotal是8，池中有8个client
    static {
        // 最大空闲数 如果是0，则每次都会获取新的连接
        // maxActive与maxIdle要保持一致
        poolConfig.maxIdle = 50;
        // 最大激活连接数
        poolConfig.maxActive = 50;
    }

    // 要池化的对象的工厂类，这个是我们要实现的类
    private static EsClientPoolFactory esClientPoolFactory = new EsClientPoolFactory();

    // 利用对象工厂类和配置类生成对象池
    private static GenericObjectPool<RestHighLevelClient> clientPool = new GenericObjectPool<>(esClientPoolFactory,
            poolConfig);

    /**
     * 获得对象
     *
     * @return
     * @throws Exception
     */
    public static RestHighLevelClient getClient() throws Exception {
        RestHighLevelClient restHighLevelClient = clientPool.borrowObject();
        logger.info("从池中取一个对象"+restHighLevelClient);
        return restHighLevelClient;
    }

    /**
     * 归还对象
     *
     * @param client
     */
    public static void returnClient(RestHighLevelClient client) throws Exception {
        logger.info("使用完毕之后，归还对象"+client);
        clientPool.returnObject(client);
    }

}

EsClientPoolFactory .java

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.pool.BasePoolableObjectFactory;
import org.apache.commons.pool.PoolableObjectFactory;
import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;

/**
 * @ClassName: EsClientPoolFactory
 * @Date: 2021/11/9
 * @Author: lvxy
 * EliasticSearch连接池工厂对象
 */
@Slf4j
public class EsClientPoolFactory extends BasePoolableObjectFactory<RestHighLevelClient> {

    /**
     * 生产对象
     */
    @Override
    public RestHighLevelClient makeObject(){
        RestHighLevelClient client = null;
        try {
            client = new RestHighLevelClient(RestClient.builder(
                    new HttpHost("192.168.111.30", 8060, "http")));

        } catch (Exception e) {
            e.printStackTrace();
        }
        log.info("对象被创建了"+client);
        return client;
    }

    /**
     * 销毁对象
     */
    @Override
    public void destroyObject(RestHighLevelClient restHighLevelClient) throws Exception {
        log.info("对象被销毁了"+restHighLevelClient);
        restHighLevelClient.close();
    }

    @Override
    public boolean validateObject(RestHighLevelClient restHighLevelClient) {
        return true;
    }

    @Override
    public void activateObject(RestHighLevelClient restHighLevelClient){
        log.info("对象被激活了"+restHighLevelClient);
    }

    @Override
    public void passivateObject(RestHighLevelClient restHighLevelClient){
        log.info("回收并执行钝化操作"+restHighLevelClient);
    }

}

Test.java

public class Test {
    public static void main(String[] args) throws InterruptedException {
        // 多线程
        for (int i = 0; i < 200; i++) {
            doTest();
        }
    }

    private static void doTest() {
        Thread thread = new Thread(
                new Runnable() {
                    @SneakyThrows
                    @Override
                    public void run() {
                        RestHighLevelClient client = ElasticSearchPoolUtil.getClient();
                        Thread.sleep(1000*3);
                        ElasticSearchPoolUtil.returnClient(client);
                    }
                }
        );
        thread.start();
    }
}

达到复用的效果
在这里插入图片描述

4 使用commons-pool2构造连接池

4.1 pom.xml

<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-pool2</artifactId>
    <version>2.9.0</version>
</dependency>

4.2 对象池类