Elasticsearch线程池

古城微笑少年丶 2021-09-17 06:32 235阅读 0赞

Elasticsearch在实际的开发使用中发现，如果每次使用都建立一次连接，那么对系统资源的消耗是巨大的，严重影响程序的性能，那么针对这种情况，我们自己实现一种线程池的，在使用时建立一次连接，然后将连接放入map中，每次使用中从map获取该连接。

具体代码如下：

/**
     * @author Shuyu.Wang
     * @package:com.ganinfo.utils
     * @className:
     * @description:ES连接池
     * @date 2018-04-18 11:26
     **/
    @Component
    @RefreshScope
    @Slf4j
    public class ClientHelper {
        @Value("${spring.data.elasticsearch.ip}")
        private String host; // 服务器地址
        @Value("${spring.data.elasticsearch.port}")
        private int port; // 端口
        @Value("${spring.data.elasticsearch.clustername}")
        private String clusterName;
    
        private Settings setting;
    
        private ConcurrentHashMap<String, TransportClient> clientMap = new ConcurrentHashMap<String, TransportClient>();
    
        private Map<String, Integer> ips = new HashMap<String, Integer>(); // hostname port
    
        public static final ClientHelper getInstance() {
            return ClientHolder.INSTANCE;
        }
    
        private static class ClientHolder {
            private static final ClientHelper INSTANCE = new ClientHelper();
        }
    
        /**
         * 初始化默认的client
         */
        public void init() {
            log.info("host:" + host);
            log.info("clusterName:" + clusterName);
            if (host.contains(",")) {
                String[] split = host.split(",");
                for (String ip : split) {
                    ips.put(ip, port);
                }
            } else {
                ips.put(host, port);
            }
            setting = Settings.builder().put("cluster.name", clusterName)
                    .put("client.transport.sniff", true)
                    .put("thread_pool.search.size", 10)
                    .put("thread_pool.search.queue_size", 500)
                    .build();
            addClient(setting, ips);
        }
    
      
    
        public TransportClient getClient() {
            return getClient(clusterName);
        }
    
        public TransportClient getClient(String clusterName) {
            if (CommonUtil.isNotMapNull(clientMap)) {
                return clientMap.get(clusterName);
            } else {
                init();
            }
            return clientMap.get(clusterName);
        }
    
        public void addClient(Settings setting, Map<String, Integer> ips) {
            TransportClient client = new PreBuiltTransportClient(setting);
    
            for (String ip : ips.keySet()) {
                try {
                    client = client.addTransportAddresses(new InetSocketTransportAddress(InetAddress.getByName(ip), Integer.valueOf(ips.get(ip).toString())));
                } catch (UnknownHostException e) {
                    e.printStackTrace();
                }
            }
            if (client != null) {
                clientMap.put(setting.get("cluster.name"), client);
            } else {
                log.error("TransportClient is null ,can not build cennection for es");
            }
    
        }
    
    
    }

使用：

@Autowired
        private ClientHelper clientHelper;
    
    
    
    
    
     @Override
        public ESReturn<List<Map<String, Object>>> query(String index, String type, QueryBuilder queryBuilder, String sort, Integer from, Integer size) {
            ESReturn<List<Map<String, Object>>> esReturn = new ESReturn<List<Map<String, Object>>>();
            TransportClient client = clientHelper.getClient();
            SearchRequestBuilder srb = client.prepareSearch(index).setTypes(type);
            SearchResponse sr = srb.setQuery(queryBuilder).setFrom(from).setSize(size).addSort(SortBuilders.fieldSort(sort).unmappedType("integer").order(SortOrder.DESC)).execute().actionGet(); // 分页排序所有
            long totalHits = sr.getHits().totalHits();
            esReturn.setTotal(totalHits);
            SearchHits hits = sr.getHits();
            List<Map<String, Object>> result = new ArrayList<>();
            for (SearchHit hit : hits) {
                Map<String, Object> map = (Map<String, Object>) hit.getSource();
                result.add(map);
            }
    //        log.info("返回结果：" + totalHits + "条数据");
            esReturn.setData(result);
            return esReturn;
        }