3

我正在尝试将 ElasticSearch REST API 与 JavaApache HttpAsyncClient库一起使用。我想使用持久流水线连接。这是一些测试代码（输出在注释中）：

@Test
public void testEsPipeliningClient() throws IOException, ExecutionException, InterruptedException
{
    testPost(HttpAsyncClients.createDefault());
    //201: {"_index":"test_index","_type":"test_type","_id":"AVIHYGnqdqqg_TAHm4ix","_version":1,"_shards":{"total":2,"successful":1,"failed":0},"created":true}
    testPost(HttpAsyncClients.createPipelining());
    //400: No handler found for uri [http://127.0.0.1:9200/test_index/test_type] and method [POST]
}

private void testPost(CloseableHttpAsyncClient client) throws ExecutionException, InterruptedException, IOException
{
    client.start();
    HttpPost request = new HttpPost("http://127.0.0.1:9200/test_index/test_type");
    request.setEntity(new StringEntity("{\"some_field\": \"some_value\"}"));
    Future<HttpResponse> responseFuture = client.execute(request, null);
    HttpResponse response = responseFuture.get();
    System.err.println(response.getStatusLine().getStatusCode() + ": " + EntityUtils.toString(response.getEntity()));
}

我不明白，为什么它适用于HttpAsyncClients.createDefault()客户端，但不适用于HttpAsyncClients.createPipelining(). 我也无法理解这两种创建方法之间的区别。

为什么我在使用时会收到错误响应createPipelining()？

我试图查看与https://httpbin.org/post的区别，但它向我展示了两个选项的相同结果。我使用默认的 ElasticSearch 设置。

谢谢！

UPD1

我尝试使用具有相同结果的PUT文档（ ）请求-它可以正常工作，但是在使用时我遇到了类似的错误-未找到处理程序<...>。PUT http://127.0.0.1/test_index/test_type/<doc id>createDefault()createPipelining()

但是当我尝试执行创建索引（PUT http://127.0.0.1/<index name>）的请求时，又出现了另一个错误。请看下面的代码：

@Test
public void testEsPipeliningClient() throws IOException, ExecutionException, InterruptedException
{
    testCreateIndex(HttpAsyncClients.createDefault());
    //200: {"acknowledged":true}
    testCreateIndex(HttpAsyncClients.createPipelining());
    //400: {"error":{"root_cause":[{"type":"mapper_parsing_exception","reason":"failed to parse, document is empty"}],"type":"mapper_parsing_exception","reason":"failed to parse, document is empty"},"status":400}
}

private void testCreateIndex(CloseableHttpAsyncClient client) throws ExecutionException, InterruptedException, IOException
{
    client.start();
    HttpPut request = new HttpPut("http://127.0.0.1:9200/" + RandomStringUtils.randomAlphabetic(8).toLowerCase());
    Future<HttpResponse> responseFuture = client.execute(request, null);
    HttpResponse response = responseFuture.get();
    System.err.println(response.getStatusLine().getStatusCode() + ": " + EntityUtils.toString(response.getEntity()));
}

正如我在此文档页面中看到的那样， ElasticSearch 默认支持 HTTP 管道。也许我需要在 ES 设置中更改什么？

UPD2

以下是UPD1部分中具有不同日志记录设置的代码的一些线路日志：

Dorg.apache.commons.logging.simplelog.log.org.apache.http=DEBUG -Dorg.apache.commons.logging.simplelog.log.org.apache.http.wire=INFO

http://pastebin.com/v29uvgbj

-Dorg.apache.commons.logging.simplelog.log.org.apache.http.impl.conn=DEBUG -Dorg.apache.commons.logging.simplelog.log.org.apache.http.impl.client=DEBUG -Dorg.apache.commons.logging.simplelog.log.org.apache.http.client=DEBUG -Dorg.apache.commons.logging.simplelog.log.org.apache.http.wire=DEBUG

http://pastebin.com/G9ij15d6

UPD3

我只是尝试用 createMinimal() 替换 createDefault()，它导致了与 createPipelining() 相同的错误。MinimalHttpAsyncClient 中的任何想法可能会导致此问题？也许有一种方法可以手动创建流水线客户端（使用构建器类）而不会出现这个问题？

4

2 回答 2

2

实际上，您只需从 URL 中提取主机并HttpPost仅使用绝对路径创建一个对象。请参阅下面第二、第三和第五行的更改：

client.start();
HttpHost targetHost = new HttpHost("127.0.0.1", 9200);
HttpPost request = new HttpPost("/test_index/test_type");
request.setEntity(new StringEntity("{\"some_field\": \"some_value\"}"));
Future<HttpResponse> responseFuture = client.execute(targetHost, request, null);
HttpResponse response = responseFuture.get();
System.out.println(response.getStatusLine().getStatusCode() + ": " + EntityUtils.toString(response.getEntity()));

进行这三个更改并再次运行代码将产生以下结果：

201: {"_index":"test_index","_type":"test_type","_id":"AVISSimIZHOoPG8ibOyF","_version":1,"created":true}
201: {"_index":"test_index","_type":"test_type","_id":"AVISSimjZHOoPG8ibOyG","_version":1,"created":true}
于 2016-01-05T14:57:17.747 回答
2

服务器必须在请求行中的绝对请求 URI 上阻塞

[DEBUG] wire - http-outgoing-1 >> "PUT http://127.0.0.1:9200/ydiwdsid HTTP/1.1[\r][\n]"

流水线模式下的 HttpAsyncClient 采用最小的协议处理链。它不会尝试重写请求对象的请求 URI。

对于您的特定情况，请求流水线似乎没有多大意义。更不用说，除非您分批提交请求，否则您甚至不会使用流水线执行。

于 2016-01-05T13:53:58.583 回答