ElasticSearch和Apache HttpAsyncClient

Question

我正在尝试将ElasticSearch REST API与Jav​​a Apache 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

我尝试了具有相同结果的PUTdocument（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中可能导致此问题？也许有一种方法可以手动创建管道客户端（带有构建器类）而不会出现此问题？

Answer 1

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

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

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

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