java - 无法在 Elastic-Search 中创建 TransportClient

Question

我一直在尝试在 Elastic-Search 中创建 TrasportClient，但不知何故无法运行它。

Java代码:

package es_example.custom;

import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.node.NodeBuilder;


import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Map;

public class EsRoutingTransportClient
{
    // Both the IP addresses are replaced with valid IP addresses in code
    String ZEN_DISCOVERY_UNICAST_HOSTS_NO_JSON = "10.100.100.100,10.111.111.111";

    String ES_PATH_HOME = "/app/devuser/elasticsearch/dummy-path-home/";

    String ES_CLUSTER_NAME = "my-cluster"; // Verified cluster exists

    Integer ES_PORT = 9200;

    private Client client;

    public EsRoutingTransportClient()
    {
        String elasticSearchHost = ZEN_DISCOVERY_UNICAST_HOSTS_NO_JSON;
        Integer defaultPort = ES_PORT;
        String elasticSearchCluster = ES_CLUSTER_NAME;

        Settings settings = Settings.settingsBuilder()
                .put("http.enabled", true)
                .put("discovery.zen.ping.multicast.enabled", false)
                // Both the IP addresses are replaced with valid IP addresses in code
                .put("discovery.zen.ping.unicast.hosts", "[\"10.100.100.100\",\"10.111.111.111\"]")
                .put("discovery.zen.minimum_master_nodes", 1)
                .put("path.home", ES_PATH_HOME)
                .build();
        client = TransportClient.builder().settings(settings).build();
        parseAndAddHostPort(elasticSearchHost, defaultPort);
    }

    private void parseAndAddHostPort(String elasticSearchHost, Integer defaultPort)
    {
        String [] esHosts = elasticSearchHost.split(",");
        for (String esHost: esHosts)
        {
            String [] hostPort = esHost.split(":");
            Integer esPort = defaultPort;

            if (hostPort.length == 2)
            {
                esHost = hostPort[0];
                esPort = Integer.parseInt(hostPort[1]);
            }
            try
            {
                client = ((TransportClient)client).addTransportAddress(
                        new InetSocketTransportAddress(InetAddress.getByName(esHost), esPort));
                System.out.println("Added " + esHost + " with port " + esPort + " as a transport address");
            } catch (UnknownHostException e) {
                e.printStackTrace();
            }
        }
    }

    public void sendMessage (String msg, String indexName, String type, String id, String routing)
    {
        try
        {
            byte[] byteBuffer = msg.getBytes();
            IndexResponse response = client.prepareIndex(indexName, type, id)
                    .setSource(byteBuffer)
                    .execute()
                    .actionGet();
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
    }

    public static void main (String args[])
    {
        EsRoutingTransportClient esClient = new EsRoutingTransportClient();
        esClient.sendMessage("Hello", "storm", "json-trips", null, null);
    }
}

pom.xml:

<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>es_example</groupId>
  <artifactId>es-proj</artifactId>
  <version>0.0.3-SNAPSHOT</version>
  <packaging>jar</packaging>
  <name>es-proj</name>
  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  </properties>
  <dependencies>
    <dependency>
      <groupId>org.elasticsearch</groupId>
      <artifactId>elasticsearch</artifactId>
      <version>2.2.0</version>
    </dependency>
  </dependencies>
  <build>
    <plugins>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-shade-plugin</artifactId>
        <version>2.4.3</version>
        <configuration />
        <executions>
          <execution>
            <phase>package</phase>
            <goals>
              <goal>shade</goal>
            </goals>
            <configuration>
              <artifactSet>
                <excludes>
                  <exlcude>com.esotericsoftware.kryo:kryo:*</exlcude>
                </excludes>
              </artifactSet>
              <transformers>
                <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                  <mainClass>es_example.custom.EsRoutingTransportClient</mainClass>
                </transformer>
              </transformers>
            </configuration>
          </execution>
        </executions>
      </plugin>
    </plugins>
  </build>
</project>

我按如下方式运行:

mvn clean package
java -jar target/es-proj-0.0.3-SNAPSHOT.jar

这是我得到的异常:

Mar 07, 2016 4:17:43 PM org.elasticsearch.plugins.PluginsService <init>
INFO: [Bullseye] modules [], plugins [], sites []
Mar 07, 2016 4:17:48 PM org.elasticsearch.client.transport.TransportClientNodesService$SimpleNodeSampler doSample
INFO: [Bullseye] failed to get node info for {#transport#-1}{10.100.100.100}{10.100.100.100:9200}, disconnecting...
ReceiveTimeoutTransportException[[][10.100.100.100:9200][cluster:monitor/nodes/liveness] request_id [0] timed out after [5003ms]]
    at org.elasticsearch.transport.TransportService$TimeoutHandler.run(TransportService.java:645)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

Added 10.100.100.100 with port 9200 as a transport address
Mar 07, 2016 4:17:53 PM org.elasticsearch.client.transport.TransportClientNodesService$SimpleNodeSampler doSample
INFO: [Bullseye] failed to get node info for {#transport#-1}{10.100.100.100}{10.100.100.100:9200}, disconnecting...
ReceiveTimeoutTransportException[[][10.100.100.100:9200][cluster:monitor/nodes/liveness] request_id [1] timed out after [5002ms]]
    at org.elasticsearch.transport.TransportService$TimeoutHandler.run(TransportService.java:645)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

Mar 07, 2016 4:17:58 PM org.elasticsearch.client.transport.TransportClientNodesService$SimpleNodeSampler doSample
INFO: [Bullseye] failed to get node info for {#transport#-1}{10.100.100.100}{10.100.100.100:9200}, disconnecting...
ReceiveTimeoutTransportException[[][10.100.100.100:9200][cluster:monitor/nodes/liveness] request_id [2] timed out after [5000ms]]
    at org.elasticsearch.transport.TransportService$TimeoutHandler.run(TransportService.java:645)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

Mar 07, 2016 4:18:03 PM org.elasticsearch.client.transport.TransportClientNodesService$SimpleNodeSampler doSample
INFO: [Bullseye] failed to get node info for {#transport#-2}{10.111.111.111}{10.111.111.111:9200}, disconnecting...
ReceiveTimeoutTransportException[[][10.111.111.111:9200][cluster:monitor/nodes/liveness] request_id [3] timed out after [5001ms]]
    at org.elasticsearch.transport.TransportService$TimeoutHandler.run(TransportService.java:645)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

Added 10.111.111.111 with port 9200 as a transport address
NoNodeAvailableException[None of the configured nodes are available: [{#transport#-1}{10.100.100.100}{10.100.100.100:9200}, {#transport#-2}{10.111.111.111}{10.111.111.111:9200}]]
    at org.elasticsearch.client.transport.TransportClientNodesService.ensureNodesAreAvailable(TransportClientNodesService.java:290)
    at org.elasticsearch.client.transport.TransportClientNodesService.execute(TransportClientNodesService.java:207)
    at org.elasticsearch.client.transport.support.TransportProxyClient.execute(TransportProxyClient.java:55)
    at org.elasticsearch.client.transport.TransportClient.doExecute(TransportClient.java:286)
    at org.elasticsearch.client.support.AbstractClient.execute(AbstractClient.java:351)
    at org.elasticsearch.action.ActionRequestBuilder.execute(ActionRequestBuilder.java:85)
    at org.elasticsearch.action.ActionRequestBuilder.execute(ActionRequestBuilder.java:59)
    at es_example.custom.EsRoutingTransportClient.sendMessage(EsRoutingTransportClient.java:82)
    at es_example.custom.EsRoutingTransportClient.main(EsRoutingTransportClient.java:94)

从同一个终端对两台主机执行 Ping 操作。

(在上面的例子中我使用了10.100.100.100和10.111.111.111。在实际代码库中，它们是真正有效的IP地址)

上面列表中提到的节点之一是主节点，另一个是数据节点。

我不知道如何解决这个问题。

任何帮助将不胜感激。

谢谢!

Answer 1

您为 TransportClient 创建的设置不正确。它们都应该是 ES 节点上的 elasticsearch.yml 文件中的设置。这些是配置 ES 服务器的设置，而不是 ES TransportClient。

因此需要在 10.100.100.100 和 10.111.111.111 上运行的节点上的 elasticsearch.yml 文件中设置以下设置

cluster.name: my-cluster
http.enabled: true
discovery.zen.ping.multicast.enabled: false
discovery.zen.ping.unicast.hosts: ["10.100.100.100","10.111.111.111"]
discovery.zen.minimum_master_nodes: 1
path.home: /path/to/es/home

此外，您还需要在每个主机上使用正确的 IP 地址进行以下设置(这是每个主机将绑定(bind)到的 IP 地址，以便其他节点和客户端可以看到它们并连接到它们):

network.host: 10.100.100.100

然后你就可以像这样构建你的 ES 客户端:

    Settings settings = Settings.settingsBuilder().build();
    client = TransportClient.builder().settings(settings).build();
    parseAndAddHostPort(elasticSearchHost, defaultPort);

这应该有效。