Elasticsearch Java Rest Client API 整理总结 (一)——Document API

时间  2019-11-11
标签 elasticsearch java rest client api 整理 总结 document 栏目 日志分析 繁體版
原文   原文链接

目录html

引言

业余时间搞 python 爬虫爬取数据，完善个人小程序；工做时间仍是要努力完成领导分配的任务，作个人 Java 老本行的。java

这不，如今就有个需求，集团要将 elasticsearch 版本从 2.2 升级到 6.3, 因为以前作项目使用 spring data es 来完成 es 数据的增删改查，如今一下升级到这么高的版本，遇到各类 API 不兼容的问题。而且 spring data es 因为总体框架 spring 等版本的限制，也不能使用了。python

无奈之下，只能使用 elasticsearch 提供的 java reset client API 来完成以前的操做。工欲善其事，必先利其器。要使用 API，第一步就是要完整，熟练的理解各个 API 的用途，限制。在学习 API 的过程当中，我将 API 的文档统一整理了一番，方便本身使用时查询，也但愿能对用到这部分的同窗提供方便。spring

注意,本 API 指南只针对 elasticsearch 6.3 版本。json

概述

Rest client 分红两部分：小程序

  • Java Low Level REST Client
    • 官方低级别 es 客户端，使用 http 协议与 Elastiicsearch 集群通讯，与全部 es 版本兼容。
  • Java High level REST Client
    • 官方高级别 es 客户端，基于低级别的客户端，它会暴露 API 特定的方法。

官方文档连接地址api

High REST Client

High Client 基于 Low Client， 主要目的是暴露一些 API，这些 API 能够接受请求对象为参数，返回响应对象，而对请求和响应细节的处理都是由 client 自动完成的。并发

每一个 API 在调用时均可以是同步或者异步的。同步和异步 API 的区别是什么呢？框架

  • 同步 API 会致使阻塞，一直等待数据返回
  • 异步 API 在命名上会加上 async 后缀，须要有一个 listener 做为参数，等这个请求返回结果或者发生错误时，这个 listener 就会被调用

起步

兼容性

  • java 1.8
  • Elasticsearch 核心项目

Java Doc 地址

只有英文版异步

Maven 配置

<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch-rest-high-level-client</artifactId>
    <version>6.3.2</version>
</dependency>

依赖

  • org.elasticsearch.client:elasticsearch-rest-client
  • org.elasticsearch:elasticsearch

初始化

RestHighLevelClient 实例依赖 REST low-level client builder

RestHighLevelClient client = new RestHighLevelClient(
        RestClient.builder(
                new HttpHost("localhost", 9200, "http"),
                new HttpHost("localhost", 9201, "http")));

High-level client 会依赖 Low-level client 来执行请求， low-level client 则会维护一个请求的线程链接池，由于当 high-level 请求处理结束时，应该 close 掉这个链接，使 low-level client 能尽快释放资源。

client.close();

文档 API

High level rest 客户端支持下面的 文档(Document) API

  • 单文档 API
    • index API
    • Get API
    • Delete API
    • Update API
  • 多文档 API
    • Bulk API
    • Multi-Get API

Index API

IndexRequest

IndexRequest request = new IndexRequest(
        "posts",  // 索引 Index
        "doc",  // Type 
        "1");  // 文档 Document Id 
String jsonString = "{" +
        "\"user\":\"kimchy\"," +
        "\"postDate\":\"2013-01-30\"," +
        "\"message\":\"trying out Elasticsearch\"" +
        "}";
request.source(jsonString, XContentType.JSON); // 文档源格式为 json string

Document Source

document source 能够是下面的格式

  • Map
Map<String, Object> jsonMap = new HashMap<>();
jsonMap.put("user", "kimchy");
jsonMap.put("postDate", new Date());
jsonMap.put("message", "trying out Elasticsearch");
IndexRequest indexRequest = new IndexRequest("posts", "doc", "1")
        .source(jsonMap);  // 会自动将 Map 转换为 JSON 格式
  • XContentBuilder : 这是 Document Source 提供的帮助类，专门用来产生 json 格式的数据
XContentBuilder builder = XContentFactory.jsonBuilder();
builder.startObject();
{
    builder.field("user", "kimchy");
    builder.timeField("postDate", new Date());
    builder.field("message", "trying out Elasticsearch");
}
builder.endObject();
IndexRequest indexRequest = new IndexRequest("posts", "doc", "1")
        .source(builder);
  • Object 键对
IndexRequest indexRequest = new IndexRequest("posts", "doc", "1")
        .source("user", "kimchy",
                "postDate", new Date(),
                "message", "trying out Elasticsearch");

同步索引

IndexResponse indexResponse = client.index(request);

异步索引

前面已经讲过，异步执行函数须要添加 listener, 而对于 index 而言，这个 listener 的类型就是 ActionListener

client.indexAsync(request, listener);

异步方法执行后会马上返回，在索引操做执行完成后，ActionListener 就会被回调:

  • 执行成功，调用 onResponse 函数
  • 执行失败，调用 onFailure 函数
ActionListener<IndexResponse> listener = new ActionListener<IndexResponse>() {
    @Override
    public void onResponse(IndexResponse indexResponse) {
        
    }

    @Override
    public void onFailure(Exception e) {
        
    }
};

IndexResponse

不论是同步回调仍是异步回调，若是调用成功，都会返回 IndexRespose 对象。 这个对象中包含什么信息呢？看下面代码

String index = indexResponse.getIndex();
String type = indexResponse.getType();
String id = indexResponse.getId();
long version = indexResponse.getVersion();
if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
   // 文档第一次建立 
} else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED) {
   // 文档以前已存在，当前是重写
}
ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
    // 成功的分片数量少于总分片数量 
}
if (shardInfo.getFailed() > 0) {
    for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
        String reason = failure.reason();  // 处理潜在的失败信息
    }
}

在索引时有版本冲突的话，会抛出 ElasticsearchException

IndexRequest request = new IndexRequest("posts", "doc", "1")
        .source("field", "value")
        .version(1); // 这里是文档版本号
try {
    IndexResponse response = client.index(request);
} catch(ElasticsearchException e) {
    if (e.status() == RestStatus.CONFLICT) {
       // 冲突了 
    }
}

若是将 opType 设置为 create, 并且若是索引的文档与已存在的文档在 index, type 和 id 上均相同，也会抛出冲突异常。

IndexRequest request = new IndexRequest("posts", "doc", "1")
        .source("field", "value")
        .opType(DocWriteRequest.OpType.CREATE);
try {
    IndexResponse response = client.index(request);
} catch(ElasticsearchException e) {
    if (e.status() == RestStatus.CONFLICT) {
        
    }
}

GET API

GET 请求

每一个 GET 请求都必须需传入下面 3 个参数

  • Index
  • Type
  • Document id
GetRequest getRequest = new GetRequest(
        "posts", 
        "doc",  
        "1");

可选参数

下面的参数都是可选的, 里面的选项并不完整，如要获取完整的属性，请参考 官方文档

  • 不获取源数据，默认是获取的
request.fetchSourceContext(FetchSourceContext.DO_NOT_FETCH_SOURCE);
  • 配置返回数据中包含指定字段
String[] includes = new String[]{"message", "*Date"};
String[] excludes = Strings.EMPTY_ARRAY;
FetchSourceContext fetchSourceContext =
        new FetchSourceContext(true, includes, excludes);
request.fetchSourceContext(fetchSourceContext);
  • 配置返回数据中排除指定字段
String[] includes = Strings.EMPTY_ARRAY;
String[] excludes = new String[]{"message"};
FetchSourceContext fetchSourceContext =
        new FetchSourceContext(true, includes, excludes);
request.fetchSourceContext(fetchSourceContext);
  • 实时 默认为 true
request.realtime(false);
  • 版本
request.version(2);
  • 版本类型
request.versionType(VersionType.EXTERNAL);

同步执行

GetResponse getResponse = client.get(getRequest);

异步执行

此部分与 index 类似, 只有一点不一样， 返回类型为 GetResponse

代码部分略

Get Response

返回的 GetResponse 对象包含要请求的文档数据（包含元数据和字段）

String index = getResponse.getIndex();
String type = getResponse.getType();
String id = getResponse.getId();
if (getResponse.isExists()) {
    long version = getResponse.getVersion();
    String sourceAsString = getResponse.getSourceAsString(); // string 形式   
    Map<String, Object> sourceAsMap = getResponse.getSourceAsMap(); // map 
    byte[] sourceAsBytes = getResponse.getSourceAsBytes(); // 字节形式 
} else {
   // 没有发现请求的文档 
}

在请求中若是包含特定的文档版本，若是与已存在的文档版本不匹配, 就会出现冲突

try {
    GetRequest request = new GetRequest("posts", "doc", "1").version(2);
    GetResponse getResponse = client.get(request);
} catch (ElasticsearchException exception) {
    if (exception.status() == RestStatus.CONFLICT) {
        // 版本冲突        
    }
}

Exists API

若是文档存在 Exists API 返回 true, 不然返回 fasle。

Exists Request

GetRequest 用法和 Get API 差很少，两个对象的可选参数是相同的。因为 exists() 方法只返回 true 或者 false， 建议将获取 _source 以及任何存储字段的值关闭，尽可能使请求轻量级。

GetRequest getRequest = new GetRequest(
    "posts",  // Index
    "doc",    // Type
    "1");     // Document id
getRequest.fetchSourceContext(new FetchSourceContext(false));  // 禁用 _source 字段
getRequest.storedFields("_none_"); // 禁止存储任何字段

同步请求

boolean exists = client.exists(getRequest);

异步请求

异步请求与 Index API 类似，此处不赘述，只粘贴代码。如需详细了解，请参阅官方地址

ActionListener<Boolean> listener = new ActionListener<Boolean>() {
    @Override
    public void onResponse(Boolean exists) {
        
    }

    @Override
    public void onFailure(Exception e) {
        
    }
};

client.existsAsync(getRequest, listener);

Delete API

官方地址

Delete Request

DeleteRequest 必须传入下面参数

DeleteRequest request = new DeleteRequest(
        "posts",   // index 
        "doc",     // doc
        "1");      // document id

可选参数

超时时间

request.timeout(TimeValue.timeValueMinutes(2)); 
request.timeout("2m");

刷新策略

request.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL); 
request.setRefreshPolicy("wait_for");

版本

request.version(2);

版本类型

request.versionType(VersionType.EXTERNAL);

同步执行

DeleteResponse deleteResponse = client.delete(request);

异步执行

ActionListener<DeleteResponse> listener = new ActionListener<DeleteResponse>() {
    @Override
    public void onResponse(DeleteResponse deleteResponse) {
        
    }

    @Override
    public void onFailure(Exception e) {
        
    }
};

client.deleteAsync(request, listener);

Delete Response

DeleteResponse 能够检索执行操做的信息，如代码所示

String index = deleteResponse.getIndex();
String type = deleteResponse.getType();
String id = deleteResponse.getId();
long version = deleteResponse.getVersion();
ReplicationResponse.ShardInfo shardInfo = deleteResponse.getShardInfo();
if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
    // 成功分片数目小于总分片
}
if (shardInfo.getFailed() > 0) {
    for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
        String reason = failure.reason(); // 处理潜在失败
    }
}

也能够来检查文档是否存在

DeleteRequest request = new DeleteRequest("posts", "doc", "does_not_exist");
DeleteResponse deleteResponse = client.delete(request);
if (deleteResponse.getResult() == DocWriteResponse.Result.NOT_FOUND) {
    // 文档不存在
}

版本冲突时也会抛出 `ElasticsearchException

try {
    DeleteRequest request = new DeleteRequest("posts", "doc", "1").version(2);
    DeleteResponse deleteResponse = client.delete(request);
} catch (ElasticsearchException exception) {
    if (exception.status() == RestStatus.CONFLICT) {
        // 版本冲突
    }
}

Update API

Update Request

UpdateRequest 的必需参数以下

UpdateRequest request = new UpdateRequest(
        "posts",  // Index
        "doc",  // 类型
        "1");   // 文档 Id

使用脚本更新

官方地址

部分文档更新

在更新部分文档时，已存在文档与部分文档会合并。

部分文档能够有如下形式

JSON 格式

UpdateRequest request = new UpdateRequest("posts", "doc", "1");
String jsonString = "{" +
        "\"updated\":\"2017-01-01\"," +
        "\"reason\":\"daily update\"" +
        "}";
request.doc(jsonString, XContentType.JSON);

Map 格式

Map<String, Object> jsonMap = new HashMap<>();
jsonMap.put("updated", new Date());
jsonMap.put("reason", "daily update");
UpdateRequest request = new UpdateRequest("posts", "doc", "1")
        .doc(jsonMap);

XContentBuilder 对象

XContentBuilder builder = XContentFactory.jsonBuilder();
builder.startObject();
{
    builder.timeField("updated", new Date());
    builder.field("reason", "daily update");
}
builder.endObject();
UpdateRequest request = new UpdateRequest("posts", "doc", "1")
        .doc(builder);

Object key-pairs

UpdateRequest request = new UpdateRequest("posts", "doc", "1")
        .doc("updated", new Date(),
             "reason", "daily update");

Upserts

若是文档不存在，可使用 upserts 方法将文档以新文档的方式建立。

UpdateRequest request = new UpdateRequest("posts", "doc", "1")
        .doc("updated", new Date(),
             "reason", "daily update");

upserts 方法支持的文档格式与 update 方法相同。

可选参数

超时时间

request.timeout(TimeValue.timeValueSeconds(1)); 
request.timeout("1s");

刷新策略

request.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL); 
request.setRefreshPolicy("wait_for");

冲突后重试次数

request.retryOnConflict(3);

获取数据源，默认是开启的

request.fetchSource(true);

包括特定字段

String[] includes = new String[]{"updated", "r*"};
String[] excludes = Strings.EMPTY_ARRAY;
request.fetchSource(new FetchSourceContext(true, includes, excludes));

排除特定字段

String[] includes = Strings.EMPTY_ARRAY;
String[] excludes = new String[]{"updated"};
request.fetchSource(new FetchSourceContext(true, includes, excludes));

指定版本

request.version(2);

禁用 noop detection

request.scriptedUpsert(true);

设置若是更新的文档不存在，就必需要建立一个

request.docAsUpsert(true);

同步执行

UpdateResponse updateResponse = client.update(request);

异步执行

此处只贴代码，官方地址

ActionListener<UpdateResponse> listener = new ActionListener<UpdateResponse>() {
    @Override
    public void onResponse(UpdateResponse updateResponse) {
        
    }

    @Override
    public void onFailure(Exception e) {
        
    }
};

client.updateAsync(request, listener);

Update Response

String index = updateResponse.getIndex();
String type = updateResponse.getType();
String id = updateResponse.getId();
long version = updateResponse.getVersion();
if (updateResponse.getResult() == DocWriteResponse.Result.CREATED) {
    // 文档已建立
} else if (updateResponse.getResult() == DocWriteResponse.Result.UPDATED) {
    // 文档已更新
} else if (updateResponse.getResult() == DocWriteResponse.Result.DELETED) {
    // 文档已删除
} else if (updateResponse.getResult() == DocWriteResponse.Result.NOOP) {
    // 文档不受更新的影响
}

若是在 UpdateRequest 中使能了获取源数据，响应中则包含了更新后的源文档信息。

GetResult result = updateResponse.getGetResult(); 
if (result.isExists()) {
    String sourceAsString = result.sourceAsString();  // 将获取的文档以 string 格式输出
    Map<String, Object> sourceAsMap = result.sourceAsMap(); // 以 Map 格式输出
    byte[] sourceAsBytes = result.source();  // 字节形式
} else {
    // 默认状况下，不会返回文档源数据
}

也能够检测是否分片失败

ReplicationResponse.ShardInfo shardInfo = updateResponse.getShardInfo();
if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
    // 成功的分片数量小于总分片数量
}
if (shardInfo.getFailed() > 0) {
    for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
        String reason = failure.reason(); // 获得分片失败的缘由
    }
}

若是在执行 UpdateRequest 时，文档不存在，响应中会包含 404 状态码，并且会抛出 ElasticsearchException 。

UpdateRequest request = new UpdateRequest("posts", "type", "does_not_exist")
        .doc("field", "value");
try {
    UpdateResponse updateResponse = client.update(request);
} catch (ElasticsearchException e) {
    if (e.status() == RestStatus.NOT_FOUND) {
        // 处理文档不存在的状况
    }
}

若是版本冲突，也会抛出 ElasticsearchException

UpdateRequest request = new UpdateRequest("posts", "doc", "1")
        .doc("field", "value")
        .version(1);
try {
    UpdateResponse updateResponse = client.update(request);
} catch(ElasticsearchException e) {
    if (e.status() == RestStatus.CONFLICT) {
        // 处理版本冲突的状况
    }
}

Bulk API 批量处理

批量请求

使用 BulkRequest 能够在一次请求中执行多个索引，更新和删除的操做。

BulkRequest request = new BulkRequest();  
request.add(new IndexRequest("posts", "doc", "1")  
        .source(XContentType.JSON,"field", "foo")); // 将第一个 IndexRequest 添加到批量请求中
request.add(new IndexRequest("posts", "doc", "2")  
        .source(XContentType.JSON,"field", "bar")); // 第二个
request.add(new IndexRequest("posts", "doc", "3")  
        .source(XContentType.JSON,"field", "baz")); // 第三个

在同一个 BulkRequest 也能够添加不一样的操做类型

BulkRequest request = new BulkRequest();
request.add(new DeleteRequest("posts", "doc", "3")); 
request.add(new UpdateRequest("posts", "doc", "2") 
        .doc(XContentType.JSON,"other", "test"));
request.add(new IndexRequest("posts", "doc", "4")  
        .source(XContentType.JSON,"field", "baz"));

可选参数

超时时间

request.timeout(TimeValue.timeValueMinutes(2)); 
request.timeout("2m");

刷新策略

request.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL); 
request.setRefreshPolicy("wait_for");

设置在批量操做前必须有几个分片处于激活状态

request.waitForActiveShards(2); 
request.waitForActiveShards(ActiveShardCount.ALL);  // 所有分片都处于激活状态
request.waitForActiveShards(ActiveShardCount.DEFAULT);  // 默认
request.waitForActiveShards(ActiveShardCount.ONE);  // 一个

同步请求

BulkResponse bulkResponse = client.bulk(request);

异步请求

与 GETAPI 等请求相似，只贴代码。

ActionListener<BulkResponse> listener = new ActionListener<BulkResponse>() {
    @Override
    public void onResponse(BulkResponse bulkResponse) {
        
    }

    @Override
    public void onFailure(Exception e) {
        
    }
};

client.bulkAsync(request, listener);

Bulk Response

BulkResponse 中包含执行操做后的信息，并容许对每一个操做结果迭代。

for (BulkItemResponse bulkItemResponse : bulkResponse) { // 遍历全部的操做结果
    DocWriteResponse itemResponse = bulkItemResponse.getResponse(); // 获取操做结果的响应，能够是  IndexResponse, UpdateResponse or DeleteResponse, 它们均可以惭怍是 DocWriteResponse 实例

    if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.INDEX
            || bulkItemResponse.getOpType() == DocWriteRequest.OpType.CREATE) { 
        IndexResponse indexResponse = (IndexResponse) itemResponse; // index 操做后的响应结果

    } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.UPDATE) { 
        UpdateResponse updateResponse = (UpdateResponse) itemResponse; // update 操做后的响应结果

    } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.DELETE) { 
        DeleteResponse deleteResponse = (DeleteResponse) itemResponse; // delete 操做后的响应结果
    }
}

此外，批量响应还有一个很是便捷的方法来检测是否有一个或多个操做失败

if (bulkResponse.hasFailures()) { 
    // 表示至少有一个操做失败
}

在这种状况下，咱们要遍历全部的操做结果，检查是不是失败的操做，并获取对应的失败信息

for (BulkItemResponse bulkItemResponse : bulkResponse) {
    if (bulkItemResponse.isFailed()) { // 检测给定的操做是否失败
        BulkItemResponse.Failure failure = bulkItemResponse.getFailure(); // 获取失败信息

    }
}

Bulk Processor

BulkProcessor 是为了简化 Bulk API 的操做提供的一个工具类，要执行操做，就须要下面组件

  • RestHighLevelClient 用来执行 BulkRequest 并获取 BulkResponse`
  • BulkProcessor.Listener 对 BulkRequest 执行先后以及失败时监听

BulkProcessor.builder 方法用来构建一个新的BulkProcessor

BulkProcessor.Listener listener = new BulkProcessor.Listener() { 
    @Override
    public void beforeBulk(long executionId, BulkRequest request) {
        // 在每一个 BulkRequest 执行前调用
    }

    @Override
    public void afterBulk(long executionId, BulkRequest request,
            BulkResponse response) {
        // 在每一个 BulkRequest 执行后调用
    }

    @Override
    public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
        // 失败时调用
    }
};

BulkProcessor bulkProcessor =
        BulkProcessor.builder(client::bulkAsync, listener).build(); // 构建 BulkProcessor, RestHighLevelClient.bulkAsync()  用来执行 BulkRequest

BulkProcessor.Builder 提供了多个方法来配置 BulkProcessor
如何来处理请求的执行。

BulkProcessor.Builder builder = BulkProcessor.builder(client::bulkAsync, listener);
builder.setBulkActions(500); // 指定多少操做时，就会刷新一次
builder.setBulkSize(new ByteSizeValue(1L, ByteSizeUnit.MB)); 
builder.setConcurrentRequests(0);  // 指定多大容量，就会刷新一次
builder.setFlushInterval(TimeValue.timeValueSeconds(10L)); // 容许并发执行的数量 
builder.setBackoffPolicy(BackoffPolicy
        .constantBackoff(TimeValue.timeValueSeconds(1L), 3));

BulkProcessor 建立后，各类请求就能够添加进去：

IndexRequest one = new IndexRequest("posts", "doc", "1").
        source(XContentType.JSON, "title",
                "In which order are my Elasticsearch queries executed?");
IndexRequest two = new IndexRequest("posts", "doc", "2")
        .source(XContentType.JSON, "title",
                "Current status and upcoming changes in Elasticsearch");
IndexRequest three = new IndexRequest("posts", "doc", "3")
        .source(XContentType.JSON, "title",
                "The Future of Federated Search in Elasticsearch");

bulkProcessor.add(one);
bulkProcessor.add(two);
bulkProcessor.add(three);

BulkProcessor 执行时，会对每一个 bulk request调用 BulkProcessor.Listener ， listener 提供了下面方法来访问 BulkRequest 和 BulkResponse:

BulkProcessor.Listener listener = new BulkProcessor.Listener() {
    @Override
    public void beforeBulk(long executionId, BulkRequest request) {
        int numberOfActions = request.numberOfActions(); // 在执行前获取操做的数量
        logger.debug("Executing bulk [{}] with {} requests",
                executionId, numberOfActions);
    }

    @Override
    public void afterBulk(long executionId, BulkRequest request,
            BulkResponse response) {
        if (response.hasFailures()) { // 执行后查看响应中是否包含失败的操做
            logger.warn("Bulk [{}] executed with failures", executionId);
        } else {
            logger.debug("Bulk [{}] completed in {} milliseconds",
                    executionId, response.getTook().getMillis());
        }
    }

    @Override
    public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
        logger.error("Failed to execute bulk", failure); // 请求失败时打印信息
    }
};

请求添加到 BulkProcessor ， 它的实例可使用下面两种方法关闭请求。

  • awaitClose() 在请求返回后或等待必定时间关闭
boolean terminated = bulkProcessor.awaitClose(30L, TimeUnit.SECONDS);
  • close() 马上关闭
bulkProcessor.close();

两个方法都会在关闭前对处理器中的请求进行刷新，并避免新的请求添加进去。

Multi-Get API

multiGet API 能够在单个 http 交互中并行的执行多个 get 请求。

Muti-Get Request

MutiGetRequest 实例化时参数为空，实例化后能够经过添加 MultiGetRequest.Item 来配置获取的信息

MultiGetRequest request = new MultiGetRequest();
request.add(new MultiGetRequest.Item(
    "index",     // 索引  
    "type",      // 类型
    "example_id"));  // 文档 id
request.add(new MultiGetRequest.Item("index", "type", "another_id"));  // 添加另一个条目

可选参数

multiGet 支持的参数与 Get API 支持的可选参数是相同的，能够在 Item 上设置它们。

官方地址

同步执行

构建 MultiGetRequest 后能够同步的方式执行multiGet

MultiGetResponse response = client.multiGet(request);

异步执行

和上面的异步执行同样，也是使用 listener 机制。

ActionListener<MultiGetResponse> listener = new ActionListener<MultiGetResponse>() {
    @Override
    public void onResponse(MultiGetResponse response) {
        
    }

    @Override
    public void onFailure(Exception e) {
        
    }
};

client.multiGetAsync(request, listener);

Multi-Get Response

MultiGetResponse 中getResponse 方法包含的 MultiGetItemResponse 顺序与请求时的相同。
MultiGetItemResponse ，若是执行成功，就会返回 GetResponse 对象，失败则返回
MultiGetResponse.Failure

MultiGetItemResponse firstItem = response.getResponses()[0];
assertNull(firstItem.getFailure());     // 执行成功，则返回 null         
GetResponse firstGet = firstItem.getResponse();  // 返回 GetResponse 对象
String index = firstItem.getIndex();
String type = firstItem.getType();
String id = firstItem.getId();
if (firstGet.isExists()) {
    long version = firstGet.getVersion();
    String sourceAsString = firstGet.getSourceAsString();  // string 格式      
    Map<String, Object> sourceAsMap = firstGet.getSourceAsMap(); // Map 
    byte[] sourceAsBytes = firstGet.getSourceAsBytes();       // bytes   
} else {
    // 没有发现文档
    // 尽管响应中会返回 404 状态码，也会返回一个有效的 GetResponse
    // 这是可使用 isExists 方法来判断
}

若是子请求中对应的 index 不存在，返回的 getFailure 方法中会包含 exception:

assertNull(missingIndexItem.getResponse());    // 获取的响应为空            
Exception e = missingIndexItem.getFailure().getFailure();  // 获取 exception
ElasticsearchException ee = (ElasticsearchException) e;    
// TODO status is broken! fix in a followup
// assertEquals(RestStatus.NOT_FOUND, ee.status());        
assertThat(e.getMessage(),
    containsString("reason=no such index"));

对版本冲突时的处理，官方说明地址

MultiGetRequest request = new MultiGetRequest();
request.add(new MultiGetRequest.Item("index", "type", "example_id")
    .version(1000L));
MultiGetResponse response = client.multiGet(request);
MultiGetItemResponse item = response.getResponses()[0];
assertNull(item.getResponse());                          
Exception e = item.getFailure().getFailure();            
ElasticsearchException ee = (ElasticsearchException) e;  
// TODO status is broken! fix in a followup
// assertEquals(RestStatus.CONFLICT, ee.status());          
assertThat(e.getMessage(),
    containsString("version conflict, current version [1] is "
        + "different than the one provided [1000]"));

结语

本文只包含 Java High level Rest Client 的 起步，和文档 API 部分，下篇文章中会包含查询 API，敬请期待。

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程