package com.artisan.test;
import com.google.gson.GsonBuilder;
import io.searchbox.client.JestClient;
import io.searchbox.client.JestClientFactory;
import io.searchbox.client.config.HttpClientConfig;
import java.util.Arrays;
import java.util.concurrent.TimeUnit;
public class Client {
    // volatile修饰，确保内存可见
    private volatile static Client client = null;
    private static JestClient jestClient;
    /**
     * 私有构造函数
     */
    private Client() {
        initJestClient(); // 初始化JestClient
    }
    /**
     * 懒汉模式
     * double Check
     * @return
     */
    public static Client getInstance() {
        if (client == null) {
            synchronized (Client.class) {
                if (client == null) {
                    client = new Client();
                }
            }
        }
        return client;
    }
    /**
     * 获取JestClient
     * @return
     */
    public static JestClient getJestClient() {
        return jestClient;
    }
    private void initJestClient() {
        // 初始化的集群节点
        String[] serverUris = new String[]{"http://127.0.0.1:9200", "http://127.0.0.1:8200"};
        JestClientFactory factory = new JestClientFactory();
        // 设置HttpClientConfig
        factory.setHttpClientConfig(new HttpClientConfig
                .Builder(Arrays.asList(serverUris))
                .discoveryEnabled(true) // 节点发现，确保访问的节点都是存活的节点，达到高可用
                .discoveryFrequency(2000, TimeUnit.MILLISECONDS) // NodeChecker的执行频率，默认10S
                .gson(new GsonBuilder().setDateFormat("yyyy-MM-dd'T'HH:mm:ss").create())
                .multiThreaded(true)
                .readTimeout(10000)
                .build());
        // 返回jestClient
        jestClient = factory.getObject();
    }
}

package com.artisan.test;
import io.searchbox.client.JestResult;
import io.searchbox.core.Get;
import java.io.IOException;
public class JestClientTest {
    /**
     * 构造函数
     */
    public JestClientTest() {
        Client.getInstance();// 初始化Client
    }
    private static void getDocumentMyStroe(String id) {
        Get get = new Get.Builder("my_store", id).type("product").build();
        JestResult result ;
        try {
            result = Client.getJestClient().execute(get);
            if (result != null) System.out.println(id + ":" + result.getJsonObject());
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    public static void main(String[] args) throws Exception {
        Thread.sleep(5000);// 先让NodeChecker运行，获取存活的节点，主线程这里先休眠5秒
        for (int i = 0; i < Integer.MAX_VALUE; i++) {
            Thread.sleep(2000);
            getDocumentMyStroe("998");
        }
    }
}

public NodeChecker(JestClient jestClient, ClientConfig clientConfig) {
       // 构建action ,可以根据前面HttpClientConfig#discoveryFilter(String discoveryFilter) 添加Node
        action = new NodesInfo.Builder()
                .withHttp()
                .addNode(clientConfig.getDiscoveryFilter())
                .build();
        this.client = jestClient;
        this.defaultScheme = clientConfig.getDefaultSchemeForDiscoveredNodes();
    // 根据discoveryFrequency(2000, TimeUnit.MILLISECONDS) 实例化一个定时任务出来 使用的Google Guava的包 
        this.scheduler = Scheduler.newFixedDelaySchedule(
                0l,
                clientConfig.getDiscoveryFrequency(),
                clientConfig.getDiscoveryFrequencyTimeUnit()
        );
       // 初始化的根节点 
        this.bootstrapServerList = ImmutableSet.copyOf(clientConfig.getServerList());
     // 实例化 discoveredServerList  为空，后续使用 
        this.discoveredServerList = new LinkedHashSet<String>();
    }

 @Override
    protected void runOneIteration() throws Exception {
        JestResult result;
        try {
            result = client.execute(action);
        } catch (CouldNotConnectException cnce) {
            // Can't connect to this node, remove it from the list
            log.error("Connect exception executing NodesInfo!", cnce);
            removeNodeAndUpdateServers(cnce.getHost());
            return;
            // do not elevate the exception since that will stop the scheduled calls.
            // throw new RuntimeException("Error executing NodesInfo!", e);
        } catch (Exception e) {
            log.error("Error executing NodesInfo!", e);
            client.setServers(bootstrapServerList);
            return;
            // do not elevate the exception since that will stop the scheduled calls.
            // throw new RuntimeException("Error executing NodesInfo!", e);
        }  
        if (result.isSucceeded()) {
            LinkedHashSet<String> httpHosts = new LinkedHashSet<String>();
            JsonObject jsonMap = result.getJsonObject();
            JsonObject nodes = (JsonObject) jsonMap.get("nodes");
            if (nodes != null) {
                for (Entry<String, JsonElement> entry : nodes.entrySet()) {
                    JsonObject host = entry.getValue().getAsJsonObject();
                    JsonElement addressElement = null;
                    if (host.has("version")) {
                        int majorVersion = Integer.parseInt(Splitter.on('.').splitToList(host.get("version").getAsString()).get(0));
                        if (majorVersion >= 5) {
                            JsonObject http = host.getAsJsonObject("http");
                            if (http != null && http.has(PUBLISH_ADDRESS_KEY_V5))
                                addressElement = http.get(PUBLISH_ADDRESS_KEY_V5);
                        }
                    }
                    if (addressElement == null) {
                        // get as a JsonElement first as some nodes in the cluster may not have an http_address
                        if (host.has(PUBLISH_ADDRESS_KEY)) addressElement = host.get(PUBLISH_ADDRESS_KEY);
                    }
                    if (addressElement != null && !addressElement.isJsonNull()) {
                        String httpAddress = getHttpAddress(addressElement.getAsString());
                        if(httpAddress != null) httpHosts.add(httpAddress);
                    }
              }
            }
            if (log.isDebugEnabled()) {
                log.debug("Discovered {} HTTP hosts: {}", httpHosts.size(), Joiner.on(',').join(httpHosts));
            }
            discoveredServerList = httpHosts;
            client.setServers(discoveredServerList);
        } else {
            log.warn("NodesInfo request resulted in error: {}", result.getErrorMessage());
            client.setServers(bootstrapServerList);
        }
    }

 /**
     * @throws IOException in case of a problem or the connection was aborted during request,
     *                     or in case of a problem while reading the response stream
     * @throws CouldNotConnectException if an {@link HttpHostConnectException} is encountered
     */
    @Override
    public <T extends JestResult> T execute(Action<T> clientRequest) throws IOException {
        return execute(clientRequest, null);
    }

public <T extends JestResult> T execute(Action<T> clientRequest, RequestConfig requestConfig) throws IOException {
      // 获取 HttpUriRequest 
        HttpUriRequest request = prepareRequest(clientRequest, requestConfig);
        CloseableHttpResponse response = null;
        try {
            response = executeRequest(request);
            return deserializeResponse(response, request, clientRequest);
        } catch (HttpHostConnectException ex) {
            throw new CouldNotConnectException(ex.getHost().toURI(), ex);
        } finally {
            if (response != null) {
                try {
                    response.close();
                } catch (IOException ex) {
                    log.error("Exception occurred while closing response stream.", ex);
                }
            }
        }
    }

HttpUriRequest request = prepareRequest(clientRequest, requestConfig);

   protected <T extends JestResult> HttpUriRequest prepareRequest(final Action<T> clientRequest, final RequestConfig requestConfig) {
        String elasticSearchRestUrl = getRequestURL(getNextServer(), clientRequest.getURI());
        HttpUriRequest request = constructHttpMethod(clientRequest.getRestMethodName(), elasticSearchRestUrl, clientRequest.getData(gson), requestConfig);
        log.debug("Request method={} url={}", clientRequest.getRestMethodName(), elasticSearchRestUrl);
        // add headers added to action
        for (Entry<String, Object> header : clientRequest.getHeaders().entrySet()) {
            request.addHeader(header.getKey(), header.getValue().toString());
        }
        return request;
    }