Elasticsearch Java Client连接池

按照Elasticsearch API，在Java端使用是ES服务需要创建Java Client，但是每一次连接都实例化一个client，对系统的消耗很大，即使在使用完毕之后将client close掉，由于服务器不能及时回收socket资源，极端情况下会导致服务器达到最大连接数。

为了解决上述问题并提高client利用率，可以参考使用池化技术复用client。

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.elasticsearch.client.Client;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.index.mapper.Mapping;
import org.elasticsearch.transport.client.PreBuiltTransportClient;

import com.chinadigitalvideo.esagent.servlet.WebServiceInit;

/**
 * Created by tgg on 16-3-17.
 */
public class ClientHelper {
    private static String ip;
    private static int port;

    private Settings setting;
    private Mapping mapping;
    
    private Map<String, Client> clientMap = new ConcurrentHashMap<String, Client>();

    private Map<String, Integer> ips = new HashMap<String, Integer>(); // hostname port

    private String clusterName = WebServiceInit.clusterName;

    private ClientHelper(String ip,Integer port) {
        init(ip,port);
        //TO-DO 添加你需要的client到helper
    }

    public static final ClientHelper getInstance(String ipConf ,Integer portConf) {
        ip=ipConf;
        port=portConf;
        return ClientHolder.INSTANCE;
    }

    private static class ClientHolder {
        private static final ClientHelper INSTANCE = new ClientHelper(ip,port);
    }

    /**
     * 初始化默认的client
     */
    public void init(String ip,int port) {
        
        ips.put(ip, port);
        setting =Settings.builder()
                .put("client.transport.sniff",true)
                .put("cluster.name",clusterName).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()) {
            addressList.add(new InetSocketTransportAddress(new InetSocketAddress(ip, ips.get(ip))));
        }
        return addressList;
    }

    public Client getClient() {
        return getClient(clusterName);
    }

    public Client getClient(String clusterName) {
        return clientMap.get(clusterName);//通过集群名称得到一个Client
    }

    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);
    }
}