2. 程式人生 > >Elasticsearch Java Rest Client API 整理總結 (一)

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

阿新 • • 發佈：2018-11-09

目錄

引言

業餘時間搞 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();

兩個方法都會在關閉前對處理器中的請求進行重新整理，並避免新的請求新增進去。

結語

本文只包含 Java High level Rest Client 的 起步，和文件 API 部分，下篇文章中會包含查詢 API，敬請期待。

相關推薦

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

目錄 引言 概述 High REST Client 起步 相容性 Java Doc 地址 Maven 配置 依賴 初始化 文件 API Index API

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

目錄 引言 業餘時間搞 python 爬蟲爬取資料，完善我的小程式；工作時間還是要努力完成領導分配的任務，做我的 Java 老本行的。 這不，現在就有個需求，集團要將 elasticsearch 版本從 2.2 升級到 6.3, 由於之前做專案使用 spring data es 來完成 es 資料的增刪改查

Elasticsearch Java Rest Client API 整理總結 (三)——Building Queries

art ear tex 基於 nts 做了 可用 per 匹配 目錄 上篇回顧 Building Queries 匹配所有的查詢 全文查詢 Full Text Queries 什麽是全文查詢？ Match 全文查詢 API 列表 基於詞項的查詢 Term Terms W

Elasticsearch Java Rest Client API 整理總結 (二) —— SearchAPI

目錄 引言 在上一篇 中主要介紹了 Document API，本節中講解 search API Search APIs Java High Level REST Client 支援下面的 Search API： Search API Search Request searchRequest 用來完成和搜尋

ElasticSearch--------------------->Elasticsearch Clients--------------------->Java REST Client API

document 開發者 href 結果 技術 clas 技術分享 例如 使用curl Elasticsearch——>Java REST Client API 概述： elasticsearch提供了REST WEB API，所以開發者可以在自己的restful

【轉】Elasticsearch Java Rest Client 指南

sta eve oat bubuko 增刪改查 cli for 在一起 min 原文地址：https://www.jianshu.com/p/d2c8326e8fa3 仔細看了下，5.6版本的RestHighLevelClient就這麽些API，有興趣的朋友可以去看

ElasticSearch學習筆記之二十九 Java REST Client

ElasticSearch學習筆記之二十九 Java REST Client Java REST Client Java High Level REST Client Compatibility(相容性) Javadoc Maven Reposi

ElasticSearch學習筆記之二十七 Java REST Client

Java REST Client Java REST Client分為2種： Java Low Level REST Client: Elasticsearch官方 low-level client . 它允許你通過http和 Elasticsearch 叢

Elasticsearch Java client（ES Client 簡介、Java REST Client、Java Client、Spring Data Elasticsearch） elasticsearch系列七：ES Java客戶端-Elasticsearch Java client（E

elasticsearch系列七：ES Java客戶端-Elasticsearch Java client（ES Client 簡介、Java REST Client、Java Client、Spring Data Elasticsearch） 一、ES Client 簡介 1. ES是一個服務

使用Java Rest Client操作Elasticsearch

Elasticsearch作為一個成熟的開源框架，對主流的多種客戶端語言都支援，比如Java，JavaScript ，PHP，.Net，Python，Ruby，CURL當然還有一些小眾的語言，雖然es官網沒支援，但是個人開發者也有一些開源的，具體的可在es官網client

Java字符串整理筆記(一)

長度 初始 1.2 不包含 string 其他 變量 pan 其中 1. 字符串 　　字符串是由許多個字符連接而成的，Java字符串可由任何字符組成，但這些字符必須包含在一對“ “” ”（雙引號）之內。Java使用java.lang包中的類來創建字符串，因此字符串是一個St

elasticsearch java 基於TransportClient API的增刪改查

返回客戶端類(基於TransportClient) @Configuration("meta") public class ConfigBeans { @Value("${es_name}") private String es_name; @V

黑馬程式設計師——Java高新技術之反射學習總結一

                                                                                    反射學習總結 基礎補充： Java程式中的各個Java類屬於同一類事務，描述這類事務的Java類名就是C

java多執行緒程式設計總結(一)

為了深入學習java多執行緒程式設計，我閱讀了高洪巖老師的《Java多執行緒程式設計核心技術》這本書，不得不說，這本書的內容實在到位，這是一本值得推薦閱讀的書，我在閱讀完電子版後還特意去圖書館找到了這本書的紙質版，於是開始了java多執行緒程式設計的學習，期間將

Java搜尋工具——Lucene例項總結(一)

    搞了一天半，終於利用lucene工具Demo完了我想要的功能，這其中包括為資料庫建立增量索引、從索引檔案根據id刪除索引、單欄位查詢功能、多欄位查詢功能、多條件查詢功能以及查詢結果關鍵字高亮顯示的功能。今天晚些的時候把這些功能進行了整理。看樣子一時半會還下不了班，就把Demo的結果 一 一 列舉下

ElasticSearch筆記整理（一）：簡介、REST與安裝配置

大數據 ElasticSearch ELK [TOC] ElasticSearch簡介 ElasticSearch是一款基於Apache Lucene構建的開源搜索引擎，它采用Java編寫並使用Lucene構建索引、提供搜索功能，ElasticSearch的目標是讓全文搜索變得簡單，開發者可以通

ElasticSearch 學習：JAVA HighLevel REST Client--獲取文件

一、按照索引號獲取文件內容 獲取文件資訊對應於HTTP的get請求，使用普通網頁請求顯示為 使用curl命令顯示為： 1.新建一個客戶client用於傳送請求和接收響應值 RestHighLevelClient client = new Res

[搜尋]ElasticSearch Java Api(一) －建立索引

ElasticSearch JAVA API 一、生成JSON 建立索引的第一步是要把物件轉換為JSON字串.官網給出了四種建立JSON文件的方法： 1.1手寫方式生成 String json = "{" + "\"user\":\"kimchy\",

Elasticsearch Query DSL 整理總結（一）—— Query DSL 概要，MatchAllQuery，全文查詢簡述

目錄 引言 雖然之前做過 elasticsearch 的專案，但是沒有對整個專案的知識點進行過系統的整理。這次趁著對 elasticsearch 版本的升級的機會（從2.2 升級到 6.3) ，又專門花時間對涉及到的知識點重新梳理了一遍。 俗話說，好記性不如爛筆頭。為了加深對 elasticsearch 的

elasticsearch java api 建立客戶端連線(Transport Client)

現在寫一個簡單的小例子建立elasticsearch客戶端連線,這裡我使用Transport Client,去建立客戶端連線 它可以應用在多個方面 在elasticsearch平臺中,可以執行建立索引,獲取索引,刪除索引,搜尋索引,在現有的叢集伺服器中在叢集伺服器中,可