import org.elasticsearch.xpack.client.PreBuiltXPackTransportClient; //导入依赖的package包/类
@Before
public void setUp() throws Exception {
    /**
     * 如果es集群安装了x-pack插件则以此种方式连接集群
     * 1. java客户端的方式是以tcp协议在9300端口上进行通信
     * 2. http客户端的方式是以http协议在9200端口上进行通信
     */
    Settings settings = Settings.builder(). put("xpack.security.user", "elastic:changeme").build();
    client = new PreBuiltXPackTransportClient(settings)
            .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("localhost"), 9300));
    elasticsearchTemplate = new ElasticsearchTemplate(client);
    final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
    credentialsProvider.setCredentials(AuthScope.ANY,
            new UsernamePasswordCredentials("elastic", "changeme"));
    restClient = RestClient.builder(new HttpHost("localhost",9200,"http"))
            .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                @Override
                public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                    return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                }
            }).build();
}
 

import org.elasticsearch.xpack.client.PreBuiltXPackTransportClient; //导入依赖的package包/类
public void testTransportClient() throws Exception {
    NodesInfoResponse nodeInfos = client().admin().cluster().prepareNodesInfo().get();
    List<NodeInfo>  nodes = nodeInfos.getNodes();
    assertTrue(nodes.size() > 0);
    TransportAddress publishAddress = randomFrom(nodes).getTransport().address().publishAddress();
    String clusterName = nodeInfos.getClusterName().value();

    Settings settings = Settings.builder()
            .put("cluster.name", clusterName)
            .put(ThreadContext.PREFIX + "." + CustomRealm.USER_HEADER, randomFrom(KNOWN_USERS))
            .put(ThreadContext.PREFIX + "." + CustomRealm.PW_HEADER, PASSWORD)
            .build();
    try (TransportClient client = new PreBuiltXPackTransportClient(settings)) {
        client.addTransportAddress(publishAddress);
        ClusterHealthResponse response = client.admin().cluster().prepareHealth().execute().actionGet();
        assertThat(response.isTimedOut(), is(false));
    }
}
 

import org.elasticsearch.xpack.client.PreBuiltXPackTransportClient; //导入依赖的package包/类
@BeforeClass
public static void startElasticsearch() throws IOException {
    // This is going to initialize our Rest Client if not initialized yet and will check
    // if security is activated on this cluster
    restClient();
    if (securityInstalled) {
        client = new PreBuiltXPackTransportClient(
                Settings.builder()
                        .put("client.transport.ignore_cluster_name", true)
                        .put("xpack.security.user", testClusterUser + ":" + testClusterPass)
                        .build()
        ).addTransportAddress(new InetSocketTransportAddress(new InetSocketAddress(testClusterHost, testClusterTransportPort)));
    } else {
        client = new PreBuiltTransportClient(Settings.builder().put("client.transport.ignore_cluster_name", true).build())
                .addTransportAddress(new InetSocketTransportAddress(new InetSocketAddress(testClusterHost, testClusterTransportPort)));
    }
}
 

import org.elasticsearch.xpack.client.PreBuiltXPackTransportClient; //导入依赖的package包/类
public void testTransportClientWrongAuthentication() throws Exception {
    NodesInfoResponse nodeInfos = client().admin().cluster().prepareNodesInfo().get();
    List<NodeInfo> nodes = nodeInfos.getNodes();
    assertTrue(nodes.size() > 0);
    TransportAddress publishAddress = randomFrom(nodes).getTransport().address().publishAddress();
    String clusterName = nodeInfos.getClusterName().value();

    Settings settings;
    if (randomBoolean()) {
        settings = Settings.builder()
                .put("cluster.name", clusterName)
                .put(ThreadContext.PREFIX + "." + CustomRealm.USER_HEADER, randomFrom(KNOWN_USERS) + randomAlphaOfLength(1))
                .put(ThreadContext.PREFIX + "." + CustomRealm.PW_HEADER, PASSWORD)
                .build();
    } else {
        settings = Settings.builder()
                .put("cluster.name", clusterName)
                .put(ThreadContext.PREFIX + "." + CustomRealm.USER_HEADER, randomFrom(KNOWN_USERS))
                .put(ThreadContext.PREFIX + "." + CustomRealm.PW_HEADER, randomAlphaOfLengthBetween(16, 32))
                .build();
    }

    try (TransportClient client = new PreBuiltXPackTransportClient(settings)) {
        client.addTransportAddress(publishAddress);
        client.admin().cluster().prepareHealth().execute().actionGet();
        fail("authentication failure should have resulted in a NoNodesAvailableException");
    } catch (NoNodeAvailableException e) {
        // expected
    }
}
 

import org.elasticsearch.xpack.client.PreBuiltXPackTransportClient; //导入依赖的package包/类
private void prepareClient() {

        Builder settingsBuilder = applySettings();
        try {
            TransportAddress ista = new TransportAddress(InetAddress.getByName(config.getElastic().getHost()), config.getElastic().getPort());
            client = new PreBuiltXPackTransportClient(settingsBuilder.build())
                    .addTransportAddress(ista);

        } catch (UnknownHostException ex) {
            logger.error(ex.getMessage(), ex);
            System.exit(-1);
        }
    }
 

import org.elasticsearch.xpack.client.PreBuiltXPackTransportClient; //导入依赖的package包/类
public void run(String[] args) {
    String host = System.getProperty("host");
    int port = Integer.parseInt(System.getProperty("port", "9343"));

    String hostBasedClusterName = host.split("\\.", 2)[0];
    String clusterName = System.getProperty("cluster", hostBasedClusterName);

    boolean enableSsl = Boolean.parseBoolean(System.getProperty("ssl", "true"));
    // Note: If enabling IPv6, then you should ensure that your host and network can route it to the Cloud endpoint.
    // (eg Docker disables IPv6 routing by default) - see also the initialization code at the top of this file.
    ip6Enabled = Boolean.parseBoolean(System.getProperty("ip6", "true"));
    ip4Enabled = Boolean.parseBoolean(System.getProperty("ip4", "true"));

    // For testing in dev environments, similar to `curl -k` option
    boolean insecure = Boolean.parseBoolean(System.getProperty("insecure", "false"));

    logger.info("Connecting to cluster: [{}] via [{}:{}] using ssl:[{}]", clusterName, host, port, enableSsl);

    // Build the settings for our client.
    Settings settings = Settings.builder()
        .put("client.transport.nodes_sampler_interval", "5s")
        .put("client.transport.sniff", false)
        .put("transport.tcp.compress", true)
        .put("cluster.name", clusterName)
        .put("xpack.security.transport.ssl.enabled", enableSsl)
        .put("request.headers.X-Found-Cluster", "${cluster.name}")
        .put("xpack.security.user", System.getProperty("xpack.security.user"))
        .put("xpack.security.transport.ssl.verification_mode", insecure ? "none" : "full")
        .build();

    // Instantiate a TransportClient and add the cluster to the list of addresses to connect to.
    // Only port 9343 (SSL-encrypted) is currently supported. The use of X-Pack security features (formerly Shield) is required.
    TransportClient client = new PreBuiltXPackTransportClient(settings);
    try {
        for (InetAddress address : InetAddress.getAllByName(host)) {
            if ((ip6Enabled && address instanceof Inet6Address)
                    || (ip4Enabled && address instanceof Inet4Address)) {
                client.addTransportAddress(new InetSocketTransportAddress(address, port));
            }
        }
    } catch (UnknownHostException e) {
        logger.error("Unable to get the host", e.getMessage());
    }

    while(true) {
        try {
            logger.info("Getting cluster health... ");
            ActionFuture<ClusterHealthResponse> healthFuture = client.admin().cluster().health(Requests.clusterHealthRequest());
            ClusterHealthResponse healthResponse = healthFuture.get(5, TimeUnit.SECONDS);
            logger.info("Got cluster health response: [{}]", healthResponse.getStatus());
        } catch(Throwable t) {
            logger.error("Unable to get cluster health response: [{}]", t.getMessage());
        }
        try {
            Thread.sleep(1000);
        } catch (InterruptedException ie) { ie.printStackTrace(); }
    }
}