對(duì)象池說明

在項(xiàng)目中宫蛆，我們經(jīng)常聽到連接池全谤，例如數(shù)據(jù)庫連接池，jedis連接池等等安皱。
apache提供了一個(gè)公共連接池pool2包提供了一個(gè)通用的對(duì)象池技術(shù)實(shí)現(xiàn)调鬓。可以很方便的基于它來實(shí)現(xiàn)自己的對(duì)象池酌伊，比如DBCP和Jedis他們的內(nèi)部對(duì)象池的實(shí)現(xiàn)就是依賴于Common-pool2腾窝。
對(duì)于有些對(duì)象來說，其創(chuàng)建的代價(jià)還是比較昂貴的，比如線程虹脯、tcp連接驴娃、數(shù)據(jù)庫連接等對(duì)象，因此對(duì)象池技術(shù)還是有其存在的意義循集。

Common-pool2由三大模塊組成：ObjectPool唇敞、PooledObject和PooledObjectFactory。

• ObjectPool：提供所有對(duì)象的存取管理暇榴。
• PooledObject：池化的對(duì)象厚棵，是對(duì)對(duì)象的一個(gè)包裝，加上了對(duì)象的一些其他信息蔼紧，包括對(duì)象的狀態(tài)（已用、空閑）狠轻，對(duì)象的創(chuàng)建時(shí)間等奸例。
• PooledObjectFactory：工廠類，負(fù)責(zé)池化對(duì)象的創(chuàng)建向楼，對(duì)象的初始化查吊，對(duì)象狀態(tài)的銷毀和對(duì)象狀態(tài)的驗(yàn)證。

另外,還有一個(gè)比較重要的GenericObjectPoolConfig

• GenericObjectPoolConfig：提供對(duì)象池的配置信息湖蜕。

使用說明

直接上案例.先看一個(gè)demo案例

@Data
@AllArgsConstructor
@NoArgsConstructor
public class Computer {
    private String name;
}

public class ComputerPoolFactory implements PooledObjectFactory<Computer> {
    public PooledObject<Computer> makeObject() {
        return new DefaultPooledObject<>(new Computer("zcj"));
    }

    public void destroyObject(PooledObject<Computer> pooledObject) {
        Computer object = pooledObject.getObject();
        object = null;
    }

    public boolean validateObject(PooledObject<Computer> pooledObject) {
        return false;
    }

    public void activateObject(PooledObject<Computer> pooledObject) {

    }

    public void passivateObject(PooledObject<Computer> pooledObject) {

    }
}

@Slf4j
public class ComputerPoolTest {

    private static GenericObjectPool<Computer> computerPool;

    public static void main(String[] args) {

        // 注冊(cè)對(duì)象池
        GenericObjectPoolConfig<Computer> poolConfig = new GenericObjectPoolConfig<>();
        poolConfig.setMaxTotal(3);
        poolConfig.setJmxEnabled(false);
        computerPool = new GenericObjectPool<>(new ComputerPoolFactory(), poolConfig);

        // 模擬多線程從對(duì)象池獲取Computer進(jìn)行業(yè)務(wù)處理
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 10; i++) {
            executorService.execute(ComputerPoolTest::doSomethingByComputer);
        }
        executorService.shutdown();
    }

    private static void doSomethingByComputer() {
        Computer computer = null;
        try {
            // 從對(duì)象池獲取對(duì)象
            computer = computerPool.borrowObject();
            System.out.println(computer);
            TimeUnit.SECONDS.sleep(1);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 使用完返回到對(duì)象池
            computerPool.returnObject(computer);
        }
    }
}

運(yùn)行結(jié)果

image.png

下面是支付寶客戶端池化案例

@Component
@Slf4j
public class AlipayClientPoolTemplate {

    // 定義池對(duì)象類型
    private GenericObjectPool<AlipayClient> pool;

    @Autowired
    private AlipayProperties alipayProperties;

    @PostConstruct
    public void init() {
        GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
        // 設(shè)置池對(duì)象數(shù)量
        poolConfig.setMaxTotal(10);
        poolConfig.setJmxEnabled(false);
        pool = new GenericObjectPool<>(new BasePooledObjectFactory<AlipayClient>() {
            @Override
            public AlipayClient create() {
                // 返回對(duì)象實(shí)例
                return new DefaultAlipayClient(alipayProperties.getServerUrl(),
                        alipayProperties.getAppid(),
                        alipayProperties.getPrivateKey(),
                        "JSON",
                        "utf-8",
                        alipayProperties.getAlipayPublicKey(),
                        "RSA2");
            }

            @Override
            public PooledObject<AlipayClient> wrap(AlipayClient obj) {
                return new DefaultPooledObject<>(obj);
            }
        }, poolConfig);
    }

    /**
     * 池獲取對(duì)象執(zhí)行方法
     */
    public <T extends AlipayResponse> T execute(AlipayRequest<T> request) {
        AlipayClient alipayClient = null;
        try {
            alipayClient = pool.borrowObject();
            T response = alipayClient.execute(request);
            log.info("調(diào)用支付寶接口,response->{},bizModel->{}", JSON.toJSONString(response), JSON.toJSONString(request.getBizModel()));
            return response;
        } catch (Exception e) {
            log.error("調(diào)用支付寶接口失敗,bizModel->{},exception:", JSON.toJSONString(request.getBizModel()), e);
            return null;
        } finally {
            if (alipayClient != null) {
                pool.returnObject(alipayClient);
            }
        }

    }


}

下面是RestTemplate的案例

池化前

@Configuration
public class RestTemplateConfig {

    @Bean
    public RestTemplate restTemplate(ClientHttpRequestFactory requestFactory){
        return new RestTemplate(requestFactory);
    }

    @Bean
    public ClientHttpRequestFactory simpleClientHttpRequestFactory(){
        SimpleClientHttpRequestFactory factory  = new SimpleClientHttpRequestFactory();
        factory.setReadTimeout(10000);
        factory.setConnectTimeout(10000);
        return factory;
    }
}

池化后

@Configuration
public class RestTemplateConfig {


    @Bean
    public RestTemplate restTemplate(ClientHttpRequestFactory requestFactory){
        return new RestTemplate(requestFactory);
    }


    @Bean
    public ClientHttpRequestFactory httpRequestFactory(){
       return new HttpComponentsClientHttpRequestFactory(httpClient());
    }

    @Bean
    public HttpClient httpClient(){
        Registry<ConnectionSocketFactory> registry = RegistryBuilder.<ConnectionSocketFactory>create()
                .register("http", PlainConnectionSocketFactory.getSocketFactory())
                .register("https", SSLConnectionSocketFactory.getSocketFactory())
                .build();

        PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager(registry);


        //設(shè)置連接池最大是500個(gè)連接
        connectionManager.setMaxTotal(500);
        //MaxPerRoute是對(duì)maxtotal的細(xì)分逻卖，每個(gè)主機(jī)的并發(fā)最大是300，route是指域名
        connectionManager.setDefaultMaxPerRoute(300);
      
        RequestConfig requestConfig = RequestConfig.custom()
                //返回?cái)?shù)據(jù)的超時(shí)時(shí)間
                .setSocketTimeout(20000)
                //連接上服務(wù)器的超時(shí)時(shí)間
                .setConnectTimeout(10000)
                //從連接池中獲取連接的超時(shí)時(shí)間
                .setConnectionRequestTimeout(1000)
                .build();
        CloseableHttpClient closeableHttpClient = HttpClientBuilder.create().setDefaultRequestConfig(requestConfig)
                .setConnectionManager(connectionManager)
                .build();

        return closeableHttpClient;
    }
}

下面是對(duì)于es的RestHighLevelClient的池化案例

es的屬性配置類EsClientParams

@Configuration
@ConfigurationProperties(prefix = "elasticsearch")
@Data
public class EsClientParams {

    private List<Map<String, String>> ips;

    public static final String HOSTNAME = "host";

    public static final String PORT = "port";

    private Integer poolSize ;
   
    public List<HttpHost> getHosts() {
        List<HttpHost> hosts = new ArrayList<>();
        for (Map<String, String> ip : ips) {
            HttpHost host = new HttpHost(ip.get(HOSTNAME), Integer.parseInt(ip.get(PORT)));
            hosts.add(host);
        }
        return hosts;
    }
}

PooledObjectFactory<RestHighLevelClient>的配置

public class EsClientPoolFactory implements PooledObjectFactory<RestHighLevelClient> {

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

    private HttpHost[] httpHosts;

    public EsClientPoolFactory(HttpHost[] httpHosts) {
        this.httpHosts = httpHosts;
    }

    /**
     * 初始化對(duì)象
     * @param arg0
     * @throws Exception
     */
    @Override
    public void activateObject(PooledObject<RestHighLevelClient> arg0){

    }
    
    /**
     * 銷毀對(duì)象
     */
    @Override
    public void destroyObject(PooledObject<RestHighLevelClient> pooledObject) throws Exception {
        RestHighLevelClient highLevelClient = pooledObject.getObject();
        highLevelClient.close();
    }

    /**
     * 生產(chǎn)對(duì)象
     */
    @Override
    public PooledObject<RestHighLevelClient> makeObject(){
        RestHighLevelClient client = null;
        try {
            client = new RestHighLevelClient(RestClient.builder(httpHosts));
 
        } catch (Exception e) {
            logger.error("es客戶端連接池創(chuàng)建新客戶端出錯(cuò)：{}", e.getMessage());
        }
        return new DefaultPooledObject<>(client);
    }

    /**
     * 對(duì)象實(shí)例返還對(duì)象池
     * @param arg0
     * @throws Exception
     */
    @Override
    public void passivateObject(PooledObject<RestHighLevelClient> arg0){
    }

    /**
     * 校驗(yàn)對(duì)象
     * @param arg0
     * @return
     */
    @Override
    public boolean validateObject(PooledObject<RestHighLevelClient> arg0) {
        return true;
    }
    
    
}

GenericObjectPool<RestHighLevelClient>的配置

@Configuration
public class EsConfiguration {

    @Autowired
    private EsClientParams esClientConfig;


    @Bean(name = "esClientPool")
    public GenericObjectPool<RestHighLevelClient> getEsClientPool() {
        return new GenericObjectPool<>(getEsClientPoolFactory(), getGenericObjectPoolConfig());
    }

    // 利用對(duì)象工廠類和配置類生成對(duì)象池
    @Bean
    public EsClientPoolFactory getEsClientPoolFactory() {
        return new EsClientPoolFactory(getHttpHostArr());
    }

    @Bean
    public GenericObjectPoolConfig getGenericObjectPoolConfig() {
        // 對(duì)象池配置類昭抒，不寫也可以评也，采用默認(rèn)配置
        GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
        poolConfig.setMaxTotal(esClientConfig.getPoolSize());
        poolConfig.setJmxEnabled(false);
        return  poolConfig;
    }

    /**
     * 構(gòu)建es連接 httpHost
     * @return
     */
    private HttpHost[] getHttpHostArr() {
        List<Map<String, String>> ips = esClientConfig.getIps();
        HttpHost[] httpHostsArr = new HttpHost[ips.size()];
        for(int i = 0; i < ips.size(); i++) {
            Map<String, String> ip  = ips.get(i);
            httpHostsArr[i] = new HttpHost(ip.get(EsClientParams.HOSTNAME), Integer.parseInt(ip.get(EsClientParams.PORT)), "http");
        }
        return httpHostsArr;
    }
}

GenericObjectPool<RestHighLevelClient>的封裝

@Component
public class ElasticSearchPool {

    // 利用對(duì)象工廠類和配置類生成對(duì)象池
    @Resource(name = "esClientPool")
    private GenericObjectPool<RestHighLevelClient> clientPool;

    /**
     * 獲得對(duì)象
     *
     * @return
     * @throws Exception
     */
    public RestHighLevelClient getClient() throws Exception {
        // 從池中取一個(gè)對(duì)象
        RestHighLevelClient client = clientPool.borrowObject();
        return client;
    }

    /**
     * 歸還對(duì)象
     *
     * @param client
     */
    public void returnClient(RestHighLevelClient client) {
        // 使用完畢之后，歸還對(duì)象
        if (clientPool != null) {
            clientPool.returnObject(client);
        }
    }

}

使用案例

@Component
public class ESClient {
    private static final Logger logger = LoggerFactory.getLogger(ESClient.class);
    @Autowired
    private ElasticSearchPool elasticSearchPool;

    public SearchResponseParser search(SearchRequest searchRequest) throws Exception {
        RestHighLevelClient client = null;
        try {
            // 對(duì)象池中獲取對(duì)象
            client = elasticSearchPool.getClient();
            logger.info("ES request：\n{}", searchRequest.toString());
            SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
            if (!RestStatus.OK.equals(searchResponse.status()) || searchResponse.isTimedOut()) {
                logger.error("ES FAIL ,response :\n{}", StringUtil.prettyJsonString(searchResponse));
                throw HsyException.of(CommonErrorEnum.REMOTE_ERROR, "es");
            }

            logger.info("ES SUCCESS,  response：\n{}", searchResponse);
            return new SearchResponseParser(searchResponse);
        } catch (Exception e) {
            logger.error("獲取esClient失敗或查詢失敗, exception:" , e);
            throw e;
        } finally {
            if (client != null) {
                elasticSearchPool.returnClient(client);
            }
        }

    }

    public SearchResponseParser scorllSearch(SearchScrollRequest searchScrollRequest) {
        SearchResponse searchResponse = null;
        RestHighLevelClient client = null;
        try {
            client = elasticSearchPool.getClient();
            logger.info("ES request：\n{}", JSONObject.toJSONString(searchScrollRequest));
            searchResponse = client.scroll(searchScrollRequest, RequestOptions.DEFAULT);
            if (!RestStatus.OK.equals(searchResponse.status()) || searchResponse.isTimedOut()) {
                logger.error(">>> es請(qǐng)求失敗 返回結(jié)果：{}", searchResponse);
            }
        } catch (Exception e) {
            logger.error("獲取esClient失敗或查詢失敗,exception:" : , e);
        } finally {
            if (client != null) {
                elasticSearchPool.returnClient(client);
            }
        }
        return new SearchResponseParser(searchResponse);
    }

    public ClearScrollResponse clearScroll(ClearScrollRequest clearScrollRequest) throws Exception {
        RestHighLevelClient client = null;
        try {
            client = elasticSearchPool.getClient();
            ClearScrollResponse clearScrollResponse = client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
            return clearScrollResponse;
        } finally {
            if (client != null) {
                elasticSearchPool.returnClient(client);
            }
        }
    }


}