上接：【ElasticSearch】从听说到了解
目录
- ES Client 简介
- 使用 Java High Level REST Client
13.1 准备工作
13.2 初始化
13.3 创建索引
13.4 添加文档
13.5 获取文档
13.6 查看文档是否存在
13.7 删除文档
13.8 更新文档
13.9 术语向量(Term Vectors)
13.10 BulkRequest 批量操作
13.11 Multi-Get Request
环境信息
语言：Java
JDK：1.8
ElasticSearch：7.2.0
12. ES Client 简介
回顾一下 ES 的架构
Elasticsearch 架构图
可以看到图中黄色的两块，分别对应了 ES 的两种连接方式。
-
Transport 连接 ，默认端口 9300
这种连接方式对应于架构图中的 Transport 这一层，这种客户端连接方式是直接连接 ES 的节点，使用 TCP 的方式进行连接 -
REST API ，默认端口 9200
这种连接方式对应于架构图中的 RESTful style API 这一层，这种客户端的连接方式是 RESTful 风格的，使用 http 的方式进行连接。其接受参数和响应均和 Transport 方式一致。
在过去的版本中，应用的比较广泛的是通过 Transport 端口调用相应的功能，但是新版本 7.x 的 ElasticSearch 已经不推荐使用 Transport 端口，而是推荐使用 REST 的方式调用功能。
ES提供了两个JAVA REST client 版本
-
Java Low Level REST Client:
低级别的REST客户端，通过 http 与集群交互，用户需自己编组请求 JSON 串，及解析响应 JSON 串。兼容所有 ES 版本。相关的 Java 接口可以参考接口文档。 -
Java High Level REST Client:
高级别的 REST 客户端从 6.0.0 开始加入，其基于低级别的 REST 客户端，目的是以 java 面向对象的方式来进行请求、响应处理，增加了编组请求 JSON 串、解析响应 JSON 串等相关 api。使用的版本需要保持和 ES 服务端的版本一致，否则会有版本问题。
每个API 支持 同步/异步 两种方式，同步方法直接返回一个结果对象。异步的方法以async为后缀，通过listener参数来通知结果。
高级 java REST 客户端依赖 Elasticsearch core project。
13. 使用 Java High Level REST Client
13.1 准备工作
- 新建一个 maven 工程
- 添加依赖
<!-- https://mvnrepository.com/artifact/org.elasticsearch.client/elasticsearch-rest-high-level-client -->
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-high-level-client</artifactId>
<version>7.2.0</version>
</dependency>
13.2 初始化
RestHighLevelClient 实例需要按如下方式构建：
RestHighLevelClient client = new RestHighLevelClient(
RestClient.builder(
new HttpHost("192.168.0.233", 9200, "http")
)
);
RestHighLevelClient 将根据提供的构建器在内部创建用于执行请求的客户端。该客户端维护一个连接池并启动一些线程，因此当您处理好请求时，应该将其关闭以释放这些资源，改操作通过调用 close() 方法来完成:
client.close();
13.3 创建索引
列出在 kibana 中的格式，方便参考
PUT /lib5
{
"settings": {
"number_of_shards": 3,
"number_of_replicas": 0
},
"mappings": {
"properties": {
"name": {
"type": "text",
"analyzer": "ik_max_word"
},
"address": {
"type": "text",
"analyzer": "ik_max_word"
}
}
}
}
- 创建索引 request 对象：索引名 lib5
request 设置参数的大多数方法，可以接收三种不同类型的对象：JSON 字符串、Map、XContent，因为 setting 参数比较简单，不适合作为示例，所以在下面设置 mapping 时说明。
CreateIndexRequest request = new CreateIndexRequest("lib5");
- 设置索引参数
request.settings(Settings.builder()
// 分片数
.put("index.number_of_shards", 3)
// 副本数
.put("index.number_of_replicas", 0));
- 设置索引的 mapping
- JSON 字符串构建
request.mapping("{\n" +
" \"properties\": {\n" +
" \"name\": {\n" +
" \"type\": \"text\",\n" +
" \"analyzer\": \"ik_max_word\"\n" +
" },\n" +
" \"address\": {\n" +
" \"type\": \"text\",\n" +
" \"analyzer\": \"ik_max_word\"\n" +
" }\n" +
" }\n" +
" }",
XContentType.JSON);
- Map 构建
Map<String, Object> map = new HashMap<>();
Map<String, Object> propMap = new HashMap<>();
map.put("properties", propMap);
Map<String, Object> nameMap = new HashMap<>();
Map<String, Object> addressMap = new HashMap<>();
propMap.put("name", nameMap);
propMap.put("address", addressMap);
nameMap.put("type", "text");
nameMap.put("analyzer", "ik_max_word");
addressMap.put("type", "text");
addressMap.put("analyzer", "ik_max_word");
request.mapping(map);
- XConttentBuilder 构建
XContentBuilder builder = XContentFactory.jsonBuilder();
builder.startObject()
.startObject("properties")
.startObject("name")
.field("type", "text")
.field("analyzer", "ik_max_word")
.endObject()
.startObject("address")
.field("type", "text")
.field("analyzer", "ik_max_word")
.endObject()
.endObject()
.endObject();
request.mapping(builder);
- 设置索引的别名
request.alias(new Alias("index5"));
5.发送请求
其他 request 发
同步方式
CreateIndexResponse createIndexResponse = client.indices().create(request, RequestOptions.DEFAULT);
异步执行
client.indices().createAsync(request, RequestOptions.DEFAULT,
new ActionListener<CreateIndexResponse>() {
@Override
public void onResponse(CreateIndexResponse response) {
// 执行成功时调用
}
@Override
public void onFailure(Exception e) {
// 执行失败时调用
}
}
);
13.4 添加文档
代码示例
RestHighLevelClient client = new RestHighLevelClient(
RestClient.builder(
new HttpHost("192.168.0.233", 9200, "http")
)
);
// 索引请求
IndexRequest request = new IndexRequest("lib5");
// 设置文档 id
request.id("1");
// 设置文档内容
String jsonString = "{\n" +
" \"name\": \"光光\",\n" +
" \"address\": \"江苏省南京市雨花台区\"\n" +
"}";
request.source(jsonString, XContentType.JSON);
try {
// 提交
client.index(request, RequestOptions.DEFAULT);
} catch (IOException e) {
e.printStackTrace();
}finally {
try {
client.close();
} catch (IOException e) {
e.printStackTrace();
}
}
request 的可选参数
request.routing("routing"); //路由值
request.timeout(TimeValue.timeValueSeconds(1)); //设置超时
request.timeout("1s"); ////以字符串形式设置超时时间
request.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL); //以WriteRequest.RefreshPolicy实例形式设置刷新策略
request.setRefreshPolicy("wait_for");//以字符串形式刷新策略
request.version(2); //文档版本
request.versionType(VersionType.EXTERNAL); //文档类型
request.opType(DocWriteRequest.OpType.CREATE); //操作类型
request.opType("create"); //操作类型 可选create或update
request.setPipeline("pipeline"); //索引文档之前要执行的摄取管道的名称
IndexResponse 对象
返回的 IndexResponse 对象允许检索关于已执行操作的信息
String index = indexResponse.getIndex();
String id = indexResponse.getId();
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();
}
}
13.5 获取文档
GetRequest getRequest = new GetRequest(
"lib5", // 索引名称
"1"); // 文档id
request 可选参数
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); //为特定字段配置源排除
request.storedFields("message"); //配置特定存储字段的检索(要求字段在映射中单独存储)
GetResponse getResponse = client.get(request, RequestOptions.DEFAULT);
String message = getResponse.getField("message").getValue(); //检索消息存储字段(要求该字段单独存储在映射中)
request.routing("routing"); //路由值
request.preference("preference"); //偏好值
request.realtime(false); //将realtime标志设置为false
request.refresh(true); //在检索文档之前执行刷新(默认为false)
request.version(2); //版本
request.versionType(VersionType.EXTERNAL); //版本类型
GetResponse 对象
返回的 GetResponse 允许检索请求的文档及其元数据和最终存储的字段。
String index = getResponse.getIndex();
String id = getResponse.getId();
if (getResponse.isExists()) {
long version = getResponse.getVersion();
String sourceAsString = getResponse.getSourceAsString(); //以字符串形式检索文档
Map<String, Object> sourceAsMap = getResponse.getSourceAsMap(); //以Map<String, Object>的形式检索文档
byte[] sourceAsBytes = getResponse.getSourceAsBytes(); //以byte[]形式检索文档
} else {
}
处理找不到文档的情况。注意，虽然返回的响应有404个状态代码，但返回的是有效的 GetResponse，而不是引发的异常。这种响应不包含任何源文档，其 isExists 方法返回 false。
当对不存在的索引执行get请求时，响应具有404状态代码，会引发一个 ElasticsearchException，需要按如下方式处理:
GetRequest request = new GetRequest("does_not_exist", "1");
try {
GetResponse getResponse = client.get(request, RequestOptions.DEFAULT);
} catch (ElasticsearchException e) {
if (e.status() == RestStatus.NOT_FOUND) {
//处理因索引不存在而引发的异常
}
}
13.6 查看文档是否存在
使用 GetRequest，就像获取应用程序接口一样。支持它的所有可选参数。由于exists()只返回true或false，建议关闭提取源和任何存储字段，这样消耗资源会少一些:
GetRequest getRequest = new GetRequest(
"posts", //索引
"1"); //文档id
getRequest.fetchSourceContext(new FetchSourceContext(false)); //禁用fetching _source.
getRequest.storedFields("_none_");
13.7 删除文档
DeleteRequest request = new DeleteRequest(
"posts", //索引
"1"); //文档id
request 可选参数
request.routing("routing"); //路由值
request.timeout(TimeValue.timeValueMinutes(2)); //以TimeValue形式设置超时
request.timeout("2m"); //以字符串形式设置超时
request.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL); //以WriteRequest.RefreshPolicy实例的形式设置刷新策略
request.setRefreshPolicy("wait_for"); //以字符串的形式设置刷新策略
request.version(2); //版本
request.versionType(VersionType.EXTERNAL); //版本类型
DeleteResponse
返回的DeleteResponse允许检索关于已执行操作的信息，如下所示:
String index = deleteResponse.getIndex();
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", "does_not_exist");
DeleteResponse deleteResponse = client.delete(
request, RequestOptions.DEFAULT);
if (deleteResponse.getResult() == DocWriteResponse.Result.NOT_FOUND) {
//如果找不到要删除的文档，执行一些操作
}
13.8 更新文档
UpdateRequest request = new UpdateRequest("posts", "1");
String jsonString = "{" +
"\"updated\":\"2017-01-01\"," +
"\"reason\":\"daily update\"" +
"}";
request.doc(jsonString, XContentType.JSON);
Upserts
如果文档尚不存在，可以使用upsert方法定义一些将作为新文档插入的内容:
String jsonString = "{\"created\":\"2017-01-01\"}";
request.upsert(jsonString, XContentType.JSON); //以字符串形式提供的Upsert文档源
request 可选参数
request.routing("routing"); //路由值
request.timeout(TimeValue.timeValueSeconds(1)); //设置超时
request.timeout("1s"); ////以字符串形式设置超时时间
request.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL); //以WriteRequest.RefreshPolicy实例形式设置刷新策略
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.setIfSeqNo(2L); //ifSeqNo
request.setIfPrimaryTerm(1L); //ifPrimaryTerm
request.detectNoop(false); //禁用noop检测
request.scriptedUpsert(true); //指出无论文档是否存在，脚本都必须运行，即如果文档不存在，脚本负责创建文档。
request.docAsUpsert(true); //指示如果部分文档尚不存在，则必须将其用作upsert文档。
request.waitForActiveShards(2); //设置在继续更新操作之前必须活动的碎片副本数量。
request.waitForActiveShards(ActiveShardCount.ALL); //ActiveShardCount的碎片副本数。可选值：ActiveShardCount.ALL, ActiveShardCount.ONE或者 ActiveShardCount.DEFAULT
Update Response
返回的更新响应允许检索关于已执行操作的信息，如下所示:
String index = updateResponse.getIndex();
String id = updateResponse.getId();
long version = updateResponse.getVersion();
if (updateResponse.getResult() == DocWriteResponse.Result.CREATED) {
//处理第一次创建文档的情况(upsert)
} else if (updateResponse.getResult() == DocWriteResponse.Result.UPDATED) {
//处理文档更新的情况
} else if (updateResponse.getResult() == DocWriteResponse.Result.DELETED) {
//处理文档被删除的情况
} else if (updateResponse.getResult() == DocWriteResponse.Result.NOOP) {
//处理文档不受更新影响的情况，即没有对文档执行任何操作(noop)
}
当通过fetchSource方法在更新请求中启用源检索时，响应包含更新文档的源:
GetResult result = updateResponse.getGetResult(); //以GetResult形式检索更新的文档
if (result.isExists()) {
String sourceAsString = result.sourceAsString(); //以字符串形式检索更新文档的来源
Map<String, Object> sourceAsMap = result.sourceAsMap(); //以Map<String, Object>的形式检索更新文档的源
byte[] sourceAsBytes = result.source(); //以byte[]的形式检索更新文档的源
} 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", "does_not_exist")
.doc("field", "value");
try {
UpdateResponse updateResponse = client.update(
request, RequestOptions.DEFAULT);
} catch (ElasticsearchException e) {
if (e.status() == RestStatus.NOT_FOUND) {
//处理由于文档不存在而引发的异常
}
}
13.9 术语向量(Term Vectors)
一个TermVectorsRequest需要一个索引和一个id来指定某个文档以及为其检索信息的字段
TermVectorsRequest request = new TermVectorsRequest("authors", "1");
request.setFields("user");
也可以为人工文档生成术语向量，即索引中不存在的文档:
XContentBuilder docBuilder = XContentFactory.jsonBuilder();
docBuilder.startObject().field("user", "guest-user").endObject();
TermVectorsRequest request = new TermVectorsRequest("authors",
docBuilder);
一个人工文档作为XContentBuilder对象提供，XContentBuilder对象是生成JSON内容的Elasticsearch内置帮助器。
可选参数
request.setFieldStatistics(false); //将字段统计设置为false(默认为true)以忽略文档统计信息。
request.setTermStatistics(true); //将术语统计设置为true(默认为false)，以显示总术语频率和文档频率。
request.setPositions(false); //将位置设置为false(默认为true)，以忽略位置输出。
request.setOffsets(false); //将偏移量设置为false(默认为true)，以忽略偏移量的输出。
request.setPayloads(false); //将有效载荷设置为false(默认为true)，以忽略有效载荷的输出。
Map<String, Integer> filterSettings = new HashMap<>();
filterSettings.put("max_num_terms", 3);
filterSettings.put("min_term_freq", 1);
filterSettings.put("max_term_freq", 10);
filterSettings.put("min_doc_freq", 1);
filterSettings.put("max_doc_freq", 100);
filterSettings.put("min_word_length", 1);
filterSettings.put("max_word_length", 10);
request.setFilterSettings(filterSettings); //设置过滤器设置，根据tf-idf分数过滤可返回的术语。
Map<String, String> perFieldAnalyzer = new HashMap<>();
perFieldAnalyzer.put("user", "keyword");
request.setPerFieldAnalyzer(perFieldAnalyzer); //将perFieldAnalyzer设置为指定与该字段不同的分析仪。
request.setRealtime(false); //将实时设置为假(默认值为真)以接近实时地检索术语向量。
request.setRouting("routing"); //设置路由参数
TermVectorsResponse
String index = response.getIndex(); //文档的索引名称。
String type = response.getType(); //文档的类型名称。
String id = response.getId(); //文档的id。
boolean found = response.getFound(); //指示是否找到文档。
检查 TermVectors
如果 TermVectorsResponse 包含非空的术语向量列表，则可以使用以下方法获得关于每个术语向量的信息:
for (TermVectorsResponse.TermVector tv : response.getTermVectorsList()) {
String fieldname = tv.getFieldName(); //当前字段的名称
int docCount = tv.getFieldStatistics().getDocCount(); //当前字段-文档计数的字段统计
long sumTotalTermFreq =
tv.getFieldStatistics().getSumTotalTermFreq(); //当前字段的字段统计信息-总术语频率之和
long sumDocFreq = tv.getFieldStatistics().getSumDocFreq(); //当前字段的字段统计信息-文档频率的总和
if (tv.getTerms() != null) {当前字段的术语
List<TermVectorsResponse.TermVector.Term> terms =
tv.getTerms(); //
for (TermVectorsResponse.TermVector.Term term : terms) {
String termStr = term.getTerm(); //术语的名称
int termFreq = term.getTermFreq(); //术语的术语频率
int docFreq = term.getDocFreq(); //记录术语的频率
long totalTermFreq = term.getTotalTermFreq(); //术语的总术语频率
float score = term.getScore(); //分数
if (term.getTokens() != null) {
List<TermVectorsResponse.TermVector.Token> tokens =
term.getTokens(); //该术语的令牌
for (TermVectorsResponse.TermVector.Token token : tokens) {
int position = token.getPosition(); //令牌的位置
int startOffset = token.getStartOffset(); //令牌的起始偏移量
int endOffset = token.getEndOffset(); //令牌的结束偏移量
String payload = token.getPayload(); //令牌的有效负载
}
}
}
}
}
13.10 BulkRequest 批量操作
BulkRequest 可用于使用单个请求执行多个索引、更新和/或删除操作。
它需要将至少一个操作添加到批量请求中:
BulkRequest request = new BulkRequest(); //创建BulkRequest
request.add(new IndexRequest("posts").id("1")
.source(XContentType.JSON,"field", "foo"));//将第一个索引请求添加到批量请求中
request.add(new IndexRequest("posts").id("2")
.source(XContentType.JSON,"field", "bar"));//添加第二个索引请求
request.add(new IndexRequest("posts").id("3")
.source(XContentType.JSON,"field", "baz"));//添加第三个索引请求
注意：Bulk API只支持JSON或SMILE中编码的文档。以任何其他格式提供文档都将导致错误。
不同的操作类型可以添加到同一个批量请求中:
BulkRequest request = new BulkRequest();
request.add(new DeleteRequest("posts", "3")); //向批量请求添加删除请求
request.add(new UpdateRequest("posts", "2")
.doc(XContentType.JSON,"other", "test"));//向批量请求添加更新请求。
request.add(new IndexRequest("posts").id("4")
.source(XContentType.JSON,"field", "baz"));//使用SMILE格式添加索引请求
可选参数
request.timeout(TimeValue.timeValueMinutes(2)); //设置超时时间
request.timeout("2m"); //设置超时时间
request.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL); //WriteRequest.RefreshPolicy实例形式设置刷新策略
request.setRefreshPolicy("wait_for"); //字符串形式设置刷新策略
request.waitForActiveShards(2); //设置在继续索引/更新/删除操作之前必须活动的碎片副本的数量。
request.waitForActiveShards(ActiveShardCount.ALL); //作为动态硬装载提供的碎片副本数，可选：ActiveShardCount.ALL, ActiveShardCount.ONE或 ActiveShardCount.DEFAULT
request.pipeline("pipelineId"); //用于所有子请求的全局管道标识
request.routing("routingId"); //用于全局路由所有子请求
BulkRequest defaulted = new BulkRequest("posts"); //具有全局索引的批量请求，用于所有子请求，除非在子请求上被重写。此参数为@Nullable，只能在批量请求创建期间设置。
BulkResponse
返回的BulkResponse包含有关已执行操作的信息，并允许迭代每个结果，如下所示:
for (BulkItemResponse bulkItemResponse : bulkResponse) { //迭代所有操作的结果
DocWriteResponse itemResponse = bulkItemResponse.getResponse(); //检索操作的响应(成功与否)，可以是索引响应、更新响应或删除响应，它们都可以被视为DocWriteResponse实例
switch (bulkItemResponse.getOpType()) {
case INDEX: //处理索引操作的响应
case CREATE:
IndexResponse indexResponse = (IndexResponse) itemResponse;
break;
case UPDATE: //处理更新操作的响应
UpdateResponse updateResponse = (UpdateResponse) itemResponse;
break;
case DELETE: //处理删除操作的响应
DeleteResponse deleteResponse = (DeleteResponse) itemResponse;
}
}
批量响应提供了一种快速检查一个或多个操作是否失败的方法:
if (bulkResponse.hasFailures()) { // 如果至少有一个操作失败，此方法返回true
}
在这种情况下，有必要迭代所有操作结果，以检查操作是否失败，如果失败，则检索相应的失败:
for (BulkItemResponse bulkItemResponse : bulkResponse) {
if (bulkItemResponse.isFailed()) { //指示给定操作是否失败
BulkItemResponse.Failure failure =
bulkItemResponse.getFailure(); //检索失败操作的失败
}
}
批量处理器
批量处理器通过提供一个实用程序类简化了Bulk API的使用，该类允许索引/更新/删除操作在添加到处理器时透明地执行。
BulkProcessor.Listener listener = new BulkProcessor.Listener() { //创建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.Builder 提供了配置批量处理器如何处理请求执行的方法:
BulkProcessor bulkProcessor = BulkProcessor.builder(
(request, bulkListener) ->
client.bulkAsync(request, RequestOptions.DEFAULT, bulkListener),
listener).build(); //通过从BulkProcessor.Builder调用build()方法来创建BulkProcessor。resthighleveloclient . BulkAsync()方法将用于在机罩下执行BulkRequest。
BulkProcessor.Builder 供了配置批量处理器如何处理请求执行的方法:
BulkProcessor.Builder builder = BulkProcessor.builder(
(request, bulkListener) ->
client.bulkAsync(request, RequestOptions.DEFAULT, bulkListener),
listener);
builder.setBulkActions(500); //根据当前添加的操作数设置刷新新批量请求的时间(默认值为1000，-1禁用)
builder.setBulkSize(new ByteSizeValue(1L, ByteSizeUnit.MB)); //根据当前添加的操作大小设置刷新新批量请求的时间(默认为5Mb，-1禁用)
builder.setConcurrentRequests(0); //设置允许执行的并发请求数(默认为1，0仅允许执行单个请求)
builder.setFlushInterval(TimeValue.timeValueSeconds(10L)); //设置一个刷新间隔，如果间隔过去，刷新任何待处理的批量请求(默认为未设置)
builder.setBackoffPolicy(BackoffPolicy
.constantBackoff(TimeValue.timeValueSeconds(1L), 3)); //设置一个恒定的后退策略，最初等待1秒钟，最多重试3次。有关更多选项，请参见BackoffPolicy.noBackoff(), BackoffPolicy.constantBackoff()和BackoffPolicy.exponentialBackoff().
一旦批量处理器被创建，可以向它添加请求:
IndexRequest one = new IndexRequest("posts").id("1")
.source(XContentType.JSON, "title",
"In which order are my Elasticsearch queries executed?");
IndexRequest two = new IndexRequest("posts").id("2")
.source(XContentType.JSON, "title",
"Current status and upcoming changes in Elasticsearch");
IndexRequest three = new IndexRequest("posts").id("3")
.source(XContentType.JSON, "title",
"The Future of Federated Search in Elasticsearch");
bulkProcessor.add(one);
bulkProcessor.add(two);
bulkProcessor.add(three);
请求将由BulkProcessor执行，它负责为每个批量请求调用BulkProcessor.Listener。
侦听器提供访问BulkRequest和BulkResponse的方法:
BulkProcessor.Listener listener = new BulkProcessor.Listener() {
@Override
public void beforeBulk(long executionId, BulkRequest request) {
int numberOfActions = request.numberOfActions(); //在每次执行BulkRequest之前调用，这个方法允许知道在BulkRequest中将要执行的操作的数量
logger.debug("Executing bulk [{}] with {} requests",
executionId, numberOfActions);
}
@Override
public void afterBulk(long executionId, BulkRequest request,
BulkResponse response) {
if (response.hasFailures()) { //在每次执行BulkRequest后调用，该方法允许知道BulkResponse是否包含错误
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); //如果BulkRequest失败，则调用该方法，该方法允许知道失败
}
};
一旦所有请求都添加到BulkProcessor中，就需要使用两种可用的关闭方法之一来关闭它的实例。
awaitClose()方法可用于等待，直到处理完所有请求或经过指定的等待时间:
//如果所有大容量请求都已完成，则该方法返回true如果在所有大容量请求完成之前等待时间已过，则该方法返回false
boolean terminated = bulkProcessor.awaitClose(30L, TimeUnit.SECONDS);
close()方法可用于立即关闭批量处理器:
bulkProcessor.close();
这两种方法都会在关闭处理器之前刷新添加到处理器中的请求，并且禁止向其中添加任何新请求。
13.11 Multi-Get Request
MultiGetRequest的构造函数为空，你可以添加MultiGetRequest.Item到查询中。
MultiGetRequest request = new MultiGetRequest();
request.add(new MultiGetRequest.Item(
"index", //索引
"example_id")); //文档id
request.add(new MultiGetRequest.Item("index", "another_id")); //添加另一个要提取的项目
可选参数
request.add(new MultiGetRequest.Item("index", "example_id")
.fetchSourceContext(FetchSourceContext.DO_NOT_FETCH_SOURCE)); //禁用源检索，默认情况下启用
String[] includes = new String[] {"foo", "*r"};
String[] excludes = Strings.EMPTY_ARRAY;
FetchSourceContext fetchSourceContext =
new FetchSourceContext(true, includes, excludes);
request.add(new MultiGetRequest.Item("index", "example_id")
.fetchSourceContext(fetchSourceContext)); //为特定字段配置源包含
String[] includes = Strings.EMPTY_ARRAY;
String[] excludes = new String[] {"foo", "*r"};
FetchSourceContext fetchSourceContext =
new FetchSourceContext(true, includes, excludes);
request.add(new MultiGetRequest.Item("index", "example_id")
.fetchSourceContext(fetchSourceContext)); //为特定字段配置源排除
request.add(new MultiGetRequest.Item("index", "example_id")
.storedFields("foo")); //配置特定存储字段的检索(要求字段在映射中单独存储)
MultiGetResponse response = client.mget(request, RequestOptions.DEFAULT);
MultiGetItemResponse item = response.getResponses()[0];
String value = item.getResponse().getField("foo").getValue(); //检索foo存储字段(要求该字段在映射中单独存储)
request.add(new MultiGetRequest.Item("index", "with_routing")
.routing("some_routing")); //路由值
request.add(new MultiGetRequest.Item("index", "with_version")
.versionType(VersionType.EXTERNAL) //版本
.version(10123L)); //版本类型
preference, realtime和refresh可以在主请求上设置，但不能在任何项目上设置:
request.preference("some_preference"); //偏好值
request.realtime(false);//将实时标志设置为false(默认true)
request.refresh(true); //在检索文档之前执行刷新(默认false)
Multi Get Response
返回的MultiGetResponse包含一个MultiGetItemResponse列表，按请求的顺序排列在GetResponse中。MultiGetResponse包含获取成功时的获取GetResponse或MultiGetResponse。失败则提示失败，成功看起来就像普通的GetResponse。
MultiGetItemResponse firstItem = response.getResponses()[0];
assertNull(firstItem.getFailure()); //getFailure返回null意味着没有失败。
GetResponse firstGet = firstItem.getResponse();//getResponse返回GetResponse。
String index = firstItem.getIndex();
String id = firstItem.getId();
if (firstGet.isExists()) {
long version = firstGet.getVersion();
String sourceAsString = firstGet.getSourceAsString(); //以字符串形式检索文档
Map<String, Object> sourceAsMap = firstGet.getSourceAsMap();//以Map<String, Object>的形式检索文档
byte[] sourceAsBytes = firstGet.getSourceAsBytes(); //以 byte[]形式检索文档
} else {
//处理找不到文档的情况。请注意，虽然返回的响应有404个状态代码，但返回的是有效的GetResponse，而不是引发的异常。这种响应不包含任何源文档，其isExists方法返回false。
}
当对不存在的索引执行的子请求之一getFailure将包含异常:
assertNull(missingIndexItem.getResponse());//getResponse为空。
Exception e = missingIndexItem.getFailure().getFailure();//getFailure不是并且包含异常。
ElasticsearchException ee = (ElasticsearchException) e; // 这个异常是一个ElasticsearchException
// TODO status is broken! fix in a followup
// assertEquals(RestStatus.NOT_FOUND, ee.status());//它的状态为“未找到”。如果不是多重获取，它应该是一个HTTP 404。
assertThat(e.getMessage(),
containsString("reason=no such index [missing_index]"));//getMessage解释了原因。
如果请求了特定的文档版本，并且现有文档具有不同的版本号，则会引发版本冲突:
MultiGetRequest request = new MultiGetRequest();
request.add(new MultiGetRequest.Item("index", "example_id")
.version(1000L));
MultiGetResponse response = client.mget(request, RequestOptions.DEFAULT);
MultiGetItemResponse item = response.getResponses()[0];
assertNull(item.getResponse());//getResponse为空。
Exception e = item.getFailure().getFailure();//getFailure不是并且包含异常。
ElasticsearchException ee = (ElasticsearchException) e;//这个异常是一个ElasticsearchException
// TODO status is broken! fix in a followup
// assertEquals(RestStatus.CONFLICT, ee.status());//它的状态为CONFLICT。如果不是多重获取，它应该是一个HTTP 409。
assertThat(e.getMessage(),
containsString("version conflict, current version [1] is "
+ "different than the one provided [1000]")); //getMessage解释了实际原因
网友评论