Elasticsearch Java Rest Client API 整理總結 (一)

来源:https://www.cnblogs.com/reycg-blog/archive/2018/11/09/9931482.html
-Advertisement-
Play Games

[TOC] 引言 業餘時間搞 python 爬蟲爬取數據，完善我的 "汽車銷量查詢小助手" ；工作時間還是要努力完成領導分配的任務，做我的 Java 老本行的。 這不，現在就有個需求，集團要將 elasticsearch 版本從 2.2 升級到 6.3, 由於之前做項目使用 來完成 es 數據的增刪 ...


目錄

引言

業餘時間搞 python 爬蟲爬取數據，完善我的小程式；工作時間還是要努力完成領導分配的任務，做我的 Java 老本行的。

這不，現在就有個需求，集團要將 elasticsearch 版本從 2.2 升級到 6.3, 由於之前做項目使用 spring data es 來完成 es 數據的增刪改查，現在一下升級到這麼高的版本，遇到各種 API 不相容的問題。並且 spring data es 由於整體框架 spring 等版本的限制，也不能使用了。

無奈之下，只能使用 elasticsearch 提供的 java reset client API 來完成之前的操作。工欲善其事，必先利其器。要使用 API，第一步就是要完整，熟練的理解各個 API 的用途，限制。在學習 API 的過程中，我將 API 的文檔統一整理了一番，方便自己使用時查詢，也希望能對用到這部分的同學提供方便。

註意,本 API 指南只針對 elasticsearch 6.3 版本。

概述

Rest client 分成兩部分：

  • Java Low Level REST Client
    • 官方低級別 es 客戶端，使用 http 協議與 Elastiicsearch 集群通信，與所有 es 版本相容。
  • Java High level REST Client
    • 官方高級別 es 客戶端，基於低級別的客戶端，它會暴露 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，敬請期待。


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • lvs的負載均衡測試
    近來工作閑暇之餘，知道自己的知識欠缺，就在網上找來一些學習視頻進行學習，在學習的時候，按照講課老師的講解步驟進行配置lvs負載均衡，配置環境如下圖： 客戶端是我本機的虛擬網卡vmnet8 lvs調度器和兩台real伺服器是vmware裝的centos7 lvs調度器配置的參數配置如下： echo 1 ...
  • 查詢字元串中子字元串出現位置
    同理，PHP中使用strpos() ...
  • JAVA數據結構
    1 數組 2 棧 3 隊列 4 鏈表 5 樹 6 堆 7 圖 8 散列表 ...
  • springboot v2.0.3版本多數據源配置
    本篇分享的是springboot多數據源配置，在從springboot v1.5版本升級到v2.0.3時，發現之前寫的多數據源的方式不可用了，捕獲錯誤信息如： 先來說下之前的多數據源配置如： 配置了兩個資料庫，在原來預設的datasource節點下麵增加了seconddatasource節點的配置， ...
  • BZOJ2763: [JLOI2011]飛行路線(分層圖 最短路)
    題意 "題目鏈接" Sol 分層圖+最短路 建$k+1$層圖，對於邊$(u, v, w)$，首先在本層內連邊權為$w$的無向邊，再各向下一層對應的節點連邊權為$0$的有向邊 如果是取最大最小值的話可以考慮二分答案+最短路 cpp // luogu judger enable o2 // luogu ...
  • Java多線程可以分組，還能這樣玩！
    前面的文章，棧長和大家分享過多線程創建的3種方式《 "實現 Java 多線程的 3 種方式" 》。 但如果線程很多的情況下，你知道如何對它們進行分組嗎？ 和 Dubbo 的服務分組一樣，Java 可以對相同性質的線程進行分組。 來看下線程類 Thread 的所有構造方法。 如圖所示，帶有 Threa ...
  • Python工作好找嗎?不懂這6個秘訣就有點難
    隨著人工智慧時代的到來，Python的需求也變的越來越大了。從目前各大IT招聘網上顯示招聘需求是非常大，而且薪資基本都在10k起，高的可達40-50k 隨著人工智慧時代的到來，Python的需求也變的越來越大了。從目前各大IT招聘網上顯示招聘需求是非常大，而且薪資基本都在10k起，高的可達40-50 ...
  • Spring MVC 學習總結（十一）——IDEA+Maven+多模塊實現SSM框架集成
    一、SSM概要 與SSH（Struts/Spring/Hibernate/）一樣，Spring+SpringMVC+MyBatis也有一個簡稱SSM，Spring實現業務對象管理，Spring MVC負責請求的轉發和視圖管理, MyBatis作為數據對象持久化引擎。這樣搭配的優點是：輕量、自由度高、 ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類