ES简单工具类ESUtil

packagecom.alibaba.otter.canal.utils;

importcom.alibaba.otter.canal.constants.ModuleEnum;
importcom.alibaba.otter.canal.custom.CanalLogUtil;
importlombok.extern.slf4j.Slf4j;
importorg.apache.commons.beanutils.BeanMap;
importorg.apache.commons.lang3.StringUtils;
importorg.apache.http.HttpHost;
importorg.elasticsearch.client.RestClient;
importorg.elasticsearch.client.RestClientBuilder;
importorg.elasticsearch.client.RestHighLevelClient;

importjava.io.IOException;
importjava.util.Arrays;
importjava.util.HashMap;
importjava.util.Map;
importjava.util.Objects;

/*** @authorkaikai_zheng
 * @version1.0.0
 * @className ESUtil
 * @description //ES工具类
 * @data 2020-07-13 11:19
 */@Slf4j
public classESUtil {

     //改为从配置中读取集群节点
//private final static String HOST_IP1 = "47.101.179.162";
//private final static String HOST_IP2 = "10.2.6.58";
//private final static String HOST_IP3 = "10.2.6.59";
//private final static String HOST_IP4 = "10.2.6.60";
//private final static int PORT = 9200;
    /*** 超时时间设为5分钟
     */
    private static final int TIME_OUT = 5 * 60 * 1000;
    private static final int ADDRESS_LENGTH = 2;
    private static final String HTTP_SCHEME = "http";

//
//private volatile static ESUtil esUtil;
//
    publicESUtil(){}

    //双重校验锁
//public static ESUtil getInstance() {
//if (null == esUtil) {
//synchronized (ESUtil.class){
//if (null == esUtil) {
//esUtil = new ESUtil();
//}
//}
//}
//return esUtil;
//}

    publicRestHighLevelClient getClient() {
        RestHighLevelClient client = newRestHighLevelClient(
                restClientBuilder()
//RestClient.builder(
//new HttpHost(HOST_IP1, PORT, PROTOCAL)
//new HttpHost(HOST_IP2, PORT, PROTOCAL),
//new HttpHost(HOST_IP3, PORT, PROTOCAL),
//new HttpHost(HOST_IP4, PORT, PROTOCAL)
);
        returnclient;
    }


    public voidcloseClient(RestHighLevelClient client) {
        try{
            client.close();
        } catch(IOException e) {
            CanalLogUtil.error(ModuleEnum.CANAL_CLIENT.getCode(), "colse es client exception,error=", e);
            e.printStackTrace();
        }
    }

    publicRestClientBuilder restClientBuilder() {
        String ipAddress[] = ResourceUtil.getResource().getProperty("es_cluster").split(",");
        CanalLogUtil.info(ModuleEnum.CANAL_CLIENT.getCode(),"load es cluster nodes result="+ipAddress);
        HttpHost[] hosts =Arrays.stream(ipAddress)
                .map(this::makeHttpHost)
                .filter(Objects::nonNull)
                .toArray(HttpHost[]::new);
        returnRestClient.builder(hosts);
    }

    privateHttpHost makeHttpHost(String s) {
        assertStringUtils.isNotEmpty(s);
        String[] address = s.split(":");
        if (address.length ==ADDRESS_LENGTH) {
            String ip = address[0];
            int port = Integer.parseInt(address[1]);
            System.err.println(ip+"+"+port);
            return newHttpHost(ip, port, HTTP_SCHEME);
        } else{
            return null;
        }
    }

    public static <T> Map<String, Object>beanToMap(T bean) {
        Map<String, Object> map = new HashMap<>();
        if (bean != null) {
            BeanMap beanMap = newBeanMap(bean);
            for(Object key : beanMap.keySet()) {
                if (beanMap.get(key) != null) {
                    map.put(key + "", beanMap.get(key));
                }
            }
        }
        returnmap;
    }

}