Elastic Search java客户端封装使用

/**
 * 
 * ESClientHelper 按照Elasticsearch API，在Java端使用是ES服务需要创建Java
 * Client，但是每一次连接都实例化一个client，对系统的消耗很大， 即使在使用完毕之后将client
 * close掉，由于服务器不能及时回收socket资源，极端情况下会导致服务器达到最大连接数。为了解决上述问题并提高client利用率，可以参考使用池化技术复用client。
 * 
 * @author xujin
 *
 */
public class ESClientHelper {
    private static final Log logger = LogFactory.getLog(ESClientHelper.class);
    private static ESClientHelper instance;
    private Settings setting;
    private Map<String, Client> clientMap = new ConcurrentHashMap<String, Client>();
    // HostName与Port
    private Map<String, Integer> ips = new HashMap<String, Integer>();
    private String clusterName = Configs.ES_CONFIG.getCluster();
    public static synchronized ESClientHelper getInstance() {
        if (instance == null) {
            instance = new ESClientHelper();
        }
        return instance;
    }
    private ESClientHelper() {
        init();
    }
    /**
     * 初始化默认的client
     */
    public void init() {
        String address = Configs.ES_CONFIG.getAddress();
        if (StringUtils.isNotEmpty(address)) {
            StringTokenizer stokenizer = new StringTokenizer(address, ",");
            while (stokenizer.hasMoreTokens()) {
                String ip = stokenizer.nextToken();
                ips.put(ip, Configs.ES_CONFIG.getPort());
            }
        }
        setting = Settings.builder().put("cluster.name", Configs.ES_CONFIG.getCluster()).build();
        addClient(setting, getAllAddress(ips));
    }
    /**
     * 获得所有的地址端口
     *
     * @return
     */
    public List<InetSocketTransportAddress> getAllAddress(Map<String, Integer> ips) {
        List<InetSocketTransportAddress> addressList = new ArrayList<InetSocketTransportAddress>();
        for (String ip : ips.keySet()) {
            try {
                addressList.add(new InetSocketTransportAddress(InetAddress.getByName(ip), ips.get(ip)));
            } catch (UnknownHostException e) {
                logger.error(" add InetSocketTransportAddress exception:[{}],ip:[{}]", e.getMessage(), ip);
            }
        }
        return addressList;
    }
    public Client getClient() {
        return getClient(clusterName);
    }
    public Client getClient(String clusterName) {
        return clientMap.get(clusterName);
    }
    public void addClient(Settings setting, List<InetSocketTransportAddress> transportAddress) {
        Client client = new PreBuiltTransportClient(setting).addTransportAddresses(
                transportAddress.toArray(new InetSocketTransportAddress[transportAddress.size()]));
        clientMap.put(setting.get("cluster.name"), client);
    }
}

Client client = TransportClient.builder().settings(setting).build().addTransportAddresses(
                transportAddress.toArray(new InetSocketTransportAddress[transportAddress.size()]));

Client client = ESClientHelper.getInstance().getClient(Configs.ES_CONFIG.getCluster());

Client client = ESClientHelper.getInstance().getClient(Configs.ES_CONFIG.getCluster());
        String index = Configs.ES_CONFIG.getIndex();
        QueryBuilder qb = matchQuery("@message.deviceInfo.deviceId", deviceId);
        SearchResponse response = client
                .prepareSearch(index).setQuery(qb)
                .setPostFilter(
                        QueryBuilders.rangeQuery("@message.apmLog.network.time.beginTime")
                                .gte(DateUtils.StringToDate(beginTime)).lte(DateUtils.StringToDate(endTime)))
                .execute().actionGet();

@Override
   public AppReqHistoryInfo getAppReqHistoryInfoByDeviceIdAndTime(String deviceId, int hourNum)
           throws ServiceException {
       String day = "now-" + String.valueOf(hourNum) + "h";
       String index = Configs.ES_CONFIG.getIndex();
       Client client = ESClientHelper.getInstance().getClient(Configs.ES_CONFIG.getCluster());
       QueryBuilder qb = matchQuery("@message.deviceInfo.deviceId", deviceId);
       MultiSearchRequestBuilder req = client.prepareMultiSearch().add(client.prepareSearch(index)
               .setPostFilter(boolQuery().must(qb).must(QueryBuilders.rangeQuery("@timestamp").gt(day).lt("now")))
               .addAggregation(AggregationBuilders.sum("REQ_SUM").field("@message.apmLog.network.reqLength"))
               .addAggregation(AggregationBuilders.sum("RSP_SUM").field("@message.apmLog.network.complete.respLength"))
               .setSize(0))
               .add(client.prepareSearch(index)
                       .setPostFilter(
                               boolQuery().must(qb).must(QueryBuilders.rangeQuery("@timestamp").gt(day).lt("now"))
                                       .must(QueryBuilders.rangeQuery("@message.apmLog.network.complete.statusCode")
                                               .gt(399).lt(999)))
                       .setSize(0))
               .add(client.prepareSearch(index).setPostFilter(
                       boolQuery().must(qb).must(QueryBuilders.rangeQuery("@timestamp").gt(day).lt("now"))
                               .must(QueryBuilders.existsQuery("@message.apmLog.network.netError.errorDomain")))
                       .setSize(0));
       AppReqHistoryInfo appReqHisInfo = new AppReqHistoryInfo();
       MultiSearchResponse rsp;
       try {
           rsp = req.execute().get();
           MultiSearchResponse.Item[] items = rsp.getResponses();
           MultiSearchResponse.Item totalItem = items[0];
           long total = totalItem.getResponse().getHits().getTotalHits();
           appReqHisInfo.setRequestNum(total);
           Double REQ_SUM = (Double) (totalItem.getResponse().getAggregations().get("REQ_SUM").getProperty("value"));
           Double RSP_SUM = (Double) (totalItem.getResponse().getAggregations().get("RSP_SUM").getProperty("value"));
           Double totalFlow = REQ_SUM + RSP_SUM;
           appReqHisInfo.setTotalFlow(totalFlow.longValue());
           MultiSearchResponse.Item statusCodeItem = items[1];
           appReqHisInfo.setHttpError(statusCodeItem.getResponse().getHits().getTotalHits());
           MultiSearchResponse.Item errorDomainItem = items[2];
           appReqHisInfo.setNetError(errorDomainItem.getResponse().getHits().getTotalHits());
       } catch (InterruptedException | ExecutionException e) {
           logger.info("getAppReqHistoryInfoByDeviceIdAndTime exception:[{}]", e.getMessage());
           throw new ServiceException("500", "getAppVisitTraceInfo exception", e.getCause());
       }
       return appReqHisInfo;
   }