Elasticsearch Java API 很全的整理
- 2019 年 10 月 17 日
- 筆記
Elasticsearch 的API 分為 REST Client API（http請求形式）以及 transportClient API兩種。相比來說transportClient API效率更高，transportClient 是通過Elasticsearch內部RPC的形式進行請求的，連接可以是一個長連接，相當於是把客戶端的請求當成
Elasticsearch 集群的一個節點。但是從Elasticsearch 7 後就會移除transportClient 。主要原因是transportClient 難以向下兼容版本。
本文中所有的講解和操作都是基於jdk 1.8 和elasticsearch 6.2.4版本。
備註：本文參考了很多Elasticsearch 的官方文檔以及部l網絡資料做的綜合整理。
一、High REST Client
High Client 基於 Low Client， 主要目的是暴露一些 API，這些 API 可以接受請求對象為參數，返迴響應對象，而對請求和響應細節的處理都是由 client 自動完成的。
API 在調用時都可以是同步或者異步兩種形式
同步 API 會導致阻塞，一直等待數據返回
異步 API 在命名上會加上 async 後綴，需要有一個 listener 作為參數，等這個請求返回結果或者發生錯誤時，這個 listener 就會被調用，listener主要是解決自動回調的問題，有點像安卓 開發裏面的listener監聽回調。
Elasticsearch REST APi 官方 地址：https://www.elastic.co/guide/en/elasticsearch/reference/6.2/index.html
Maven 依賴
<dependency>
<groupId>org.elasticsearch</groupId>
<artifactId>elasticsearch</artifactId>
<version>6.2.4</version>
</dependency>
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-high-level-client</artifactId>
<version>6.2.4</version>
</dependency>
client初始化：
RestHighLevelClient 實例依賴 REST low-level client builder
public class ElasticSearchClient {
private String[] hostsAndPorts;
public ElasticSearchClient(String[] hostsAndPorts) {
this.hostsAndPorts = hostsAndPorts;
}
public RestHighLevelClient getClient() { RestHighLevelClient client = null; List<HttpHost> httpHosts = new ArrayList<HttpHost>(); if (hostsAndPorts.length > 0) { for (String hostsAndPort : hostsAndPorts) { String[] hp = hostsAndPort.split(":"); httpHosts.add(new HttpHost(hp[0], Integer.valueOf(hp[1]), "http")); } client = new RestHighLevelClient( RestClient.builder(httpHosts.toArray(new HttpHost[0]))); } else { client = new RestHighLevelClient( RestClient.builder(new HttpHost("127.0.0.1", 9200, "http"))); } return client; }
}
文檔 API（High level rest 客戶端支持下面的 文檔(Document) API）：
- 單文檔 API：
- index API
- Get API
- Delete API
- Update API
- 多文檔 API：
- Bulk API
- Multi-Get API
1、Index API：
IndexRequest：
封裝好的參考方法：
private IndexRequest getIndexRequest(String index, String indexType, String docId, Map<String, Object> dataMap) { IndexRequest indexRequest = null; if (null == index || null == indexType) { throw new ElasticsearchException("index or indexType must not be null"); } if (null == docId) { indexRequest = new IndexRequest(index, indexType); } else { indexRequest = new IndexRequest(index, indexType, docId); } return indexRequest; } /** * 同步執行索引 * * @param index * @param indexType * @param docId * @param dataMap * @throws IOException */ public IndexResponse execIndex(String index, String indexType, String docId, Map<String, Object> dataMap) throws IOException { return getClient().index(getIndexRequest(index, indexType, docId, dataMap).source(dataMap)); } /** * 異步執行 * * @param index * @param indexType * @param docId * @param dataMap * @param indexResponseActionListener * @throws IOException */ public void asyncExecIndex(String index, String indexType, String docId, Map<String, Object> dataMap, ActionListener<IndexResponse> indexResponseActionListener) throws IOException { getClient().indexAsync(getIndexRequest(index, indexType, docId, dataMap).source(dataMap), indexResponseActionListener); }
API解釋：
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) { } }
2、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) { // 版本衝突 } }
封裝好的參考方法： /** * @param index * @param indexType * @param docId * @param includes 返回需要包含的字段，可以傳入空 * @param excludes 返回需要不包含的字段，可以傳入為空 * @param excludes version * @param excludes versionType * @return * @throws IOException */ public GetResponse getRequest(String index, String indexType, String docId, String[] includes, String[] excludes, Integer version, VersionType versionType) throws IOException { if (null == includes || includes.length == 0) { includes = Strings.EMPTY_ARRAY; } if (null == excludes || excludes.length == 0) { excludes = Strings.EMPTY_ARRAY; } GetRequest getRequest = new GetRequest(index, indexType, docId); FetchSourceContext fetchSourceContext = new FetchSourceContext(true, includes, excludes); getRequest.realtime(true); if (null != version) { getRequest.version(version); } if (null != versionType) { getRequest.versionType(versionType); } return getClient().get(getRequest.fetchSourceContext(fetchSourceContext)); } /** * @param index * @param indexType * @param docId * @param includes * @param excludes * @return * @throws IOException */ public GetResponse getRequest(String index, String indexType, String docId, String[] includes, String[] excludes) throws IOException { return getRequest(index, indexType, docId, includes, excludes, null, null); } /** * @param index * @param indexType * @param docId * @return * @throws IOException */ public GetResponse getRequest(String index, String indexType, String docId) throws IOException { GetRequest getRequest = new GetRequest(index, indexType, docId); return getClient().get(getRequest); }
3、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);
封裝的參考方法：
/** * @param index * @param indexType * @param docId * @return * @throws IOException */ public Boolean existDoc(String index, String indexType, String docId) throws IOException { GetRequest getRequest = new GetRequest(index, indexType, docId); getRequest.fetchSourceContext(new FetchSourceContext(false)); getRequest.storedFields("_none_"); return getClient().exists(getRequest); }
4、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) { // 版本衝突 } }
封裝好的參考方法：
/** * @param index * @param indexType * @param docId * @param timeValue * @param refreshPolicy * @param version * @param versionType * @return * @throws IOException */ public DeleteResponse deleteDoc(String index, String indexType, String docId, TimeValue timeValue, WriteRequest.RefreshPolicy refreshPolicy, Integer version, VersionType versionType) throws IOException { DeleteRequest deleteRequest = new DeleteRequest(index, indexType, docId); if (null != timeValue) { deleteRequest.timeout(timeValue); } if (null != refreshPolicy) { deleteRequest.setRefreshPolicy(refreshPolicy); } if (null != version) { deleteRequest.version(version); } if (null != versionType) { deleteRequest.versionType(versionType); } return getClient().delete(deleteRequest); } /** * @param index * @param indexType * @param docId * @return * @throws IOException */ public DeleteResponse deleteDoc(String index, String indexType, String docId) throws IOException { return deleteDoc(index, indexType, docId, null, null, null, null); }
5、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) { // 處理版本衝突的情況 } }
封裝好的參考方法：
/** * @param index * @param indexType * @param docId * @param dataMap * @param timeValue * @param refreshPolicy * @param version * @param versionType * @param docAsUpsert * @param includes * @param excludes * @return * @throws IOException */ public UpdateResponse updateDoc(String index, String indexType, String docId, Map<String, Object> dataMap, TimeValue timeValue, WriteRequest.RefreshPolicy refreshPolicy, Integer version, VersionType versionType, Boolean docAsUpsert, String[] includes, String[] excludes) throws IOException { UpdateRequest updateRequest = new UpdateRequest(index, indexType, docId); updateRequest.doc(dataMap); if (null != timeValue) { updateRequest.timeout(timeValue); } if (null != refreshPolicy) { updateRequest.setRefreshPolicy(refreshPolicy); } if (null != version) { updateRequest.version(version); } if (null != versionType) { updateRequest.versionType(versionType); } updateRequest.docAsUpsert(docAsUpsert); //衝突時重試的次數 updateRequest.retryOnConflict(3); if (null == includes && null == excludes) { return getClient().update(updateRequest); } else { if (null == includes || includes.length == 0) { includes = Strings.EMPTY_ARRAY; } if (null == excludes || excludes.length == 0) { excludes = Strings.EMPTY_ARRAY; } return getClient().update(updateRequest.fetchSource(new FetchSourceContext(true, includes, excludes))); } } /** * 更新時不存在就插入 * * @param index * @param indexType * @param docId * @param dataMap * @return * @throws IOException */ public UpdateResponse upDdateocAsUpsert(String index, String indexType, String docId, Map<String, Object> dataMap) throws IOException { return updateDoc(index, indexType, docId, dataMap, null, null, null, null, true, null, null); } /** * 存在才更新 * * @param index * @param indexType * @param docId * @param dataMap * @return * @throws IOException */ public UpdateResponse updateDoc(String index, String indexType, String docId, Map<String, Object> dataMap) throws IOException { return updateDoc(index, indexType, docId, dataMap, null, null, null, null, false, null, null); }
6、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);
異步請求
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.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();
兩個方法都會在關閉前對處理器中的請求進行刷新，並避免新的請求添加進去。
封裝好的參考方法：
/** * 批量操作 * * @param indexBeanList * @param timeValue * @param refreshPolicy * @return * @throws IOException */ public BulkResponse bulkRequest(List<IndexBean> indexBeanList, TimeValue timeValue, WriteRequest.RefreshPolicy refreshPolicy) throws IOException { BulkRequest bulkRequest = getBulkRequest(indexBeanList); if (null != timeValue) { bulkRequest.timeout(timeValue); } if (null != refreshPolicy) { bulkRequest.setRefreshPolicy(refreshPolicy); } return getClient().bulk(bulkRequest); } private BulkRequest getBulkRequest(List<IndexBean> indexBeanList) { BulkRequest bulkRequest = new BulkRequest(); indexBeanList.forEach(indexBean -> { if ("1".equals(indexBean.getOperateType())) { bulkRequest.add(null != indexBean.getDocId() ? new IndexRequest(indexBean.getIndex(), indexBean.getIndexType(), indexBean.getDocId()) : new IndexRequest(indexBean.getIndex(), indexBean.getIndexType())); } else if ("2".equals(indexBean.getOperateType())) { if ((null != indexBean.getDocId())) { throw new ElasticsearchException("update action docId must not be null"); } bulkRequest.add(new UpdateRequest(indexBean.getIndex(), indexBean.getIndexType(), indexBean.getDocId())); } else if ("3".equals(indexBean.getOperateType())) { if ((null != indexBean.getDocId())) { throw new ElasticsearchException("delete action docId must not be null"); } bulkRequest.add(new DeleteRequest(indexBean.getIndex(), indexBean.getIndexType(), indexBean.getDocId())); } else { throw new ElasticsearchException("OperateType" + indexBean.getOperateType() + "is not support"); } }); return bulkRequest; } /** * 批量操作 * * @param indexBeanList * @return */ public BulkResponse bulkRequest(List<IndexBean> indexBeanList) throws IOException { return bulkRequest(indexBeanList, null, null); } /** * 批量異步操作 * * @param indexBeanList * @param bulkResponseActionListener */ public void AsyncBulkRequest(List<IndexBean> indexBeanList, ActionListener<BulkResponse> bulkResponseActionListener) { getClient().bulkAsync(getBulkRequest(indexBeanList), bulkResponseActionListener); }
7、Search APIs：
Java High Level REST Client 支持下面的 Search API：
- Search API
- Search Scroll API
- Clear Scroll API
- Multi-Search API
- Ranking Evaluation API
Search API
Search Request
searchRequest 用來完成和搜索文檔，聚合，建議等相關的任何操作同時也提供了各種方式來完成對查詢結果的高亮操作。
最基本的查詢操作如下
SearchRequest searchRequest = new SearchRequest(); SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder(); searchSourceBuilder.query(QueryBuilders.matchAllQuery()); // 添加 match_all 查詢 searchRequest.source(searchSourceBuilder); // 將 SearchSourceBuilder 添加到 SeachRequest 中
可選參數
SearchRequest searchRequest = new SearchRequest("posts"); // 設置搜索的 index searchRequest.types("doc"); // 設置搜索的 type
除了配置 index 和 type 外，還有一些其他的可選參數
searchRequest.routing("routing"); // 設置 routing 參數 searchRequest.preference("_local"); // 配置搜索時偏愛使用本地分片，默認是使用隨機分片
什麼是 routing 參數?
當索引一個文檔的時候，文檔會被存儲在一個主分片上。在存儲時一般都會有多個主分片。Elasticsearch 如何知道一個文檔應該放置在哪個分片呢？這個過程是根據下面的這個公式來決定的：
shard = hash(routing) % number_of_primary_shards
routing 是一個可變值，默認是文檔的 _id ,也可以設置成一個自定義的值
number_of_primary_shards 是主分片數量
所有的文檔 API 都接受一個叫做 routing 的路由參數，通過這個參數我們可以自定義文檔到分片的映射。一個自定義的路由參數可以用來確保所有相關的文檔——例如所有屬於同一個用戶的文檔——都被存儲到同一個分片中。
使用 SearchSourceBuilder
對搜索行為的配置可以使用 SearchSourceBuilder 來完成，來看一個實例
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); // 默認配置 sourceBuilder.query(QueryBuilders.termQuery("user", "kimchy")); // 設置搜索，可以是任何類型的 QueryBuilder sourceBuilder.from(0); // 起始 index sourceBuilder.size(5); // 大小 size sourceBuilder.timeout(new TimeValue(60, TimeUnit.SECONDS)); // 設置搜索的超時時間
設置完成後，就可以添加到 SearchRequest 中。
SearchRequest searchRequest = new SearchRequest(); searchRequest.source(sourceBuilder);
構建查詢條件
查詢請求是通過使用 QueryBuilder 對象來完成的，並且支持 Query DSL。
DSL (domain-specific language) 領域特定語言，是指專註於某個應用程序領域的計算機語言。
可以使用構造函數來創建 QueryBuilder
MatchQueryBuilder matchQueryBuilder = new MatchQueryBuilder("user", "kimchy");
QueryBuilder 創建後，就可以調用方法來配置它的查詢選項：
matchQueryBuilder.fuzziness(Fuzziness.AUTO); // 模糊查詢 matchQueryBuilder.prefixLength(3); // 前綴查詢的長度 matchQueryBuilder.maxExpansions(10); // max expansion 選項，用來控制模糊查詢
也可以使用QueryBuilders 工具類來創建 QueryBuilder 對象。這個類提供了函數式編程風格的各種方法用來快速創建 QueryBuilder 對象。
QueryBuilder matchQueryBuilder = QueryBuilders.matchQuery("user", "kimchy") .fuzziness(Fuzziness.AUTO) .prefixLength(3) .maxExpansions(10);
fuzzy-matching 拼寫錯誤時的匹配：
好的全文檢索不應該是完全相同的限定邏輯，相反，可以擴大範圍來包括可能的匹配，從而根據相關性得分將更好的匹配放在前面。
例如，搜索 quick brown fox 時會匹配一個包含 fast brown foxes 的文檔
不論什麼方式創建的 QueryBuilder ，最後都需要添加到 `SearchSourceBuilder 中
searchSourceBuilder.query(matchQueryBuilder);
構建查詢 文檔中提供了一個豐富的查詢列表，裏面包含各種查詢對應的QueryBuilder 對象以及QueryBuilder helper 方法，大家可以去參考。
關於構建查詢的內容會在下篇文章中講解，敬請期待。
指定排序
SearchSourceBuilder 允許添加一個或多個SortBuilder 實例。這裡包含 4 種特殊的實現, (Field-, Score-, GeoDistance- 和 ScriptSortBuilder)
sourceBuilder.sort(new ScoreSortBuilder().order(SortOrder.DESC)); // 根據分數 _score 降序排列 (默認行為) sourceBuilder.sort(new FieldSortBuilder("_uid").order(SortOrder.ASC)); // 根據 id 降序排列
過濾數據源
默認情況下，查詢請求會返迴文檔的內容 _source ,當然我們也可以配置它。例如，禁止對 _source 的獲取
sourceBuilder.fetchSource(false);
也可以使用通配符模式以更細的粒度包含或排除特定的字段：
String[] includeFields = new String[] {"title", "user", "innerObject.*"}; String[] excludeFields = new String[] {"_type"}; sourceBuilder.fetchSource(includeFields, excludeFields);
高亮請求
可以通過在 SearchSourceBuilder 上設置 HighlightBuilder 完成對結果的高亮，而且可以配置不同的字段具有不同的高亮行為。
SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder(); HighlightBuilder highlightBuilder = new HighlightBuilder(); HighlightBuilder.Field highlightTitle = new HighlightBuilder.Field("title"); // title 字段高亮 highlightTitle.highlighterType("unified"); // 配置高亮類型 highlightBuilder.field(highlightTitle); // 添加到 builder HighlightBuilder.Field highlightUser = new HighlightBuilder.Field("user"); highlightBuilder.field(highlightUser); searchSourceBuilder.highlighter(highlightBuilder);
聚合請求
要實現聚合請求分兩步
創建合適的 `AggregationBuilder
作為參數配置在 `SearchSourceBuilder 上
SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder(); TermsAggregationBuilder aggregation = AggregationBuilders.terms("by_company") .field("company.keyword"); aggregation.subAggregation(AggregationBuilders.avg("average_age") .field("age")); searchSourceBuilder.aggregation(aggregation);
建議請求 Requesting Suggestions
SuggestionBuilder 實現類是由 SuggestBuilders 工廠類來創建的。
SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder(); SuggestionBuilder termSuggestionBuilder = SuggestBuilders.termSuggestion("user").text("kmichy"); SuggestBuilder suggestBuilder = new SuggestBuilder(); suggestBuilder.addSuggestion("suggest_user", termSuggestionBuilder); searchSourceBuilder.suggest(suggestBuilder);
對請求和聚合分析
分析 API 可用來對一個特定的查詢操作中的請求和聚合進行分析，此時要將SearchSourceBuilder 的 profile標誌位設置為 true
SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder(); searchSourceBuilder.profile(true);
只要 SearchRequest 執行完成，對應的 SearchResponse 響應中就會包含 分析結果
同步執行
同步執行是阻塞式的，只有結果返回後才能繼續執行。
SearchResponse searchResponse = client.search(searchRequest);
異步執行
異步執行使用的是 listener 對結果進行處理。
ActionListener<SearchResponse> listener = new ActionListener<SearchResponse>() { @Override public void onResponse(SearchResponse searchResponse) { // 查詢成功 } @Override public void onFailure(Exception e) { // 查詢失敗 } };
SearchResponse
查詢執行完成後，會返回 SearchResponse 對象，並在對象中包含查詢執行的細節和符合條件的文檔集合。
歸納一下， SerchResponse 包含的信息如下
請求本身的信息，如 HTTP 狀態碼，執行時間，或者請求是否超時
RestStatus status = searchResponse.status(); // HTTP 狀態碼 TimeValue took = searchResponse.getTook(); // 查詢佔用的時間 Boolean terminatedEarly = searchResponse.isTerminatedEarly(); // 是否由於 SearchSourceBuilder 中設置 terminateAfter 而過早終止 boolean timedOut = searchResponse.isTimedOut(); // 是否超時
查詢影響的分片數量的統計信息，成功和失敗的分片
int totalShards = searchResponse.getTotalShards(); int successfulShards = searchResponse.getSuccessfulShards(); int failedShards = searchResponse.getFailedShards(); for (ShardSearchFailure failure : searchResponse.getShardFailures()) { // failures should be handled here }
檢索 SearchHits
要訪問返回的文檔，首先要在響應中獲取其中的 SearchHits
SearchHits hits = searchResponse.getHits();
SearchHits 中包含了所有命中的全局信息，如查詢命中的數量或者最大分值：
long totalHits = hits.getTotalHits(); float maxScore = hits.getMaxScore();
查詢的結果嵌套在 SearchHits 中，可以通過遍歷循環獲取
SearchHit[] searchHits = hits.getHits(); for (SearchHit hit : searchHits) { // do something with the SearchHit }
SearchHit 提供了如 index ， type， docId 和每個命中查詢的分數
String index = hit.getIndex(); String type = hit.getType(); String id = hit.getId(); float score = hit.getScore();
而且，還可以獲取到文檔的源數據，以 JSON-String 形式或者 key-value map 對的形式。在 map 中，字段可以是普通類型，或者是列表類型，嵌套對象。
String sourceAsString = hit.getSourceAsString(); Map<String, Object> sourceAsMap = hit.getSourceAsMap(); String documentTitle = (String) sourceAsMap.get("title"); List<Object> users = (List<Object>) sourceAsMap.get("user"); Map<String, Object> innerObject = (Map<String, Object>) sourceAsMap.get("innerObject");
Search API 查詢關係
上面的 QueryBuilder ， SearchSourceBuilder 和 SearchRequest 之間都是嵌套關係， 可以參考下圖：
8、全文查詢 Full Text Queries
什麼是全文查詢？
像使用 match 或者 query_string 這樣的高層查詢都屬於全文查詢，
查詢 日期（date） 或整數（integer） 字段，會將查詢字符串分別作為日期或整數對待。
查詢一個（ not_analyzed ）未分析的精確值字符串字段，會將整個查詢字符串作為單個詞項對待。
查詢一個（ analyzed ）已分析的全文字段，會先將查詢字符串傳遞到一個合適的分析器，然後生成一個供查詢的詞項列表
組成了詞項列表，後面就會對每個詞項逐一執行底層查詢，將查詢結果合併，並且為每個文檔生成最終的相關度評分。
Match
match 查詢的單個詞的步驟是什麼？
檢查字段類型，查看字段是 analyzed, not_analyzed
分析查詢字符串，如果只有一個單詞項， match 查詢在執行時就會是單個底層的 term 查詢
查找匹配的文檔，會在倒排索引中查找匹配文檔，然後獲取一組包含該項的文檔
為每個文檔評分
構建 Match 查詢
match 查詢可以接受 text/numeric/dates 格式的參數，分析，並構建一個查詢。
GET /_search { "query": { "match" : { "message" : "this is a test" } } }
上面的實例中 message 是一個字段名。
對應的 QueryBuilder class : MatchQueryBuilder
具體方法 : QueryBuilders.matchQuery()
全文查詢 API 列表
基於詞項的查詢
這種類型的查詢不需要分析，它們是對單個詞項操作，只是在倒排索引中查找準確的詞項（精確匹配）並且使用 TF/IDF 算法為每個包含詞項的文檔計算相關度評分 _score。
Term
term 查詢可用作精確值匹配，精確值的類型則可以是數字，時間，布爾類型，或者是那些 not_analyzed 的字符串。
對應的 QueryBuilder class 是TermQueryBuilder
具體方法是 QueryBuilders.termQuery()
Terms
terms 查詢允許指定多個值進行匹配。如果這個字段包含了指定值中的任何一個值，就表示該文檔滿足條件。
對應的 QueryBuilder class 是 TermsQueryBuilder
具體方法是 QueryBuilders.termsQuery()
Wildcard
wildcard 通配符查詢是一種底層基於詞的查詢，它允許指定匹配的正則表達式。而且它使用的是標準的 shell 通配符查詢：
? 匹配任意字符
* 匹配 0 個或多個字符
wildcard 需要掃描倒排索引中的詞列表才能找到所有匹配的詞，然後依次獲取每個詞相關的文檔 ID。
由於通配符和正則表達式只能在查詢時才能完成，因此查詢效率會比較低，在需要高性能的場合，應當謹慎使用。
對應的 QueryBuilder class 是 WildcardQueryBuilder
具體方法是 QueryBuilders.wildcardQuery()
基於詞項 API 列表
複合查詢
什麼是複合查詢？
複合查詢會將其他的複合查詢或者葉查詢包裹起來，以嵌套的形式展示和執行，得到的結果也是對各個子查詢結果和分數的合併。可以分為下面幾種：
constant_score query
經常用在使用 filter 的場合，所有匹配的文檔分數都是一個不變的常量
bool query
可以將多個葉查詢和組合查詢再組合起來，可接受的參數如下
must : 文檔必須匹配這些條件才能被包含進來
must_not 文檔必須不匹配才能被包含進來
should 如果滿足其中的任何語句，都會增加分數；即使不滿足，也沒有影響
filter 以過濾模式進行，不評分，但是必須匹配
dis_max query
叫做分離最大化查詢，它會將任何與查詢匹配的文檔都作為結果返回，但是只是將其中最佳匹配的評分作為最終的評分返回。
function_score query
允許為每個與主查詢匹配的文檔應用一個函數，可用來改變甚至替換原始的評分
boosting query
用來控制（提高或降低）複合查詢中子查詢的權重。
特殊查詢
Wrapper Query
這裡比較重要的一個是 Wrapper Query，是說可以接受任何其他 base64 編碼的字符串作為子查詢。
主要應用場合就是在 Rest High-Level REST client 中接受 json 字符串作為參數。比如使用 gson 等 json 庫將要查詢的語句拼接好，直接塞到 Wrapper Query 中查詢就可以了，非常方便。
Wrapper Query 對應的 QueryBuilder class 是WrapperQueryBuilder
具體方法是 QueryBuilders.wrapperQuery()
9、關於 REST Client的完整工具類代碼
public class IndexBean { //index name private String index; //index type private String indexType; //index doc id private String docId; // 1 IndexRequest 2 UpdateRequest 3 DeleteRequest private String operateType; public String getOperateType() { return operateType; } public void setOperateType(String operateType) { this.operateType = operateType; } public String getIndex() { return index; } public void setIndex(String index) { this.index = index; } public String getIndexType() { return indexType; } public void setIndexType(String indexType) { this.indexType = indexType; } public String getDocId() { return docId; } public void setDocId(String docId) { this.docId = docId; } }
/** * 自定義的es異常類 */ public class ElasticsearchException extends RuntimeException { public ElasticsearchException(String s, Exception e) { super(s, e); } public ElasticsearchException(String s){ super(s); } }
import org.apache.http.HttpHost; import org.elasticsearch.action.ActionListener; import org.elasticsearch.action.bulk.BulkRequest; import org.elasticsearch.action.bulk.BulkResponse; import org.elasticsearch.action.delete.DeleteRequest; import org.elasticsearch.action.delete.DeleteResponse; import org.elasticsearch.action.get.GetRequest; import org.elasticsearch.action.get.GetResponse; import org.elasticsearch.action.index.IndexRequest; import org.elasticsearch.action.index.IndexResponse; import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.action.support.WriteRequest; import org.elasticsearch.action.update.UpdateRequest; import org.elasticsearch.action.update.UpdateResponse; import org.elasticsearch.client.RestClient; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.common.Strings; import org.elasticsearch.common.unit.TimeValue; import org.elasticsearch.index.VersionType; import org.elasticsearch.index.query.MatchQueryBuilder; import org.elasticsearch.index.query.TermQueryBuilder; import org.elasticsearch.search.builder.SearchSourceBuilder; import org.elasticsearch.search.fetch.subphase.FetchSourceContext; import org.elasticsearch.search.sort.SortBuilder; import java.io.IOException; import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; /** * es操作 * */ public class ElasticSearchClient { private String[] hostsAndPorts; public ElasticSearchClient(String[] hostsAndPorts) { this.hostsAndPorts = hostsAndPorts; } public RestHighLevelClient getClient() { RestHighLevelClient client = null; List<HttpHost> httpHosts = new ArrayList<HttpHost>(); if (hostsAndPorts.length > 0) { for (String hostsAndPort : hostsAndPorts) { String[] hp = hostsAndPort.split(":"); httpHosts.add(new HttpHost(hp[0], Integer.valueOf(hp[1]), "http")); } client = new RestHighLevelClient( RestClient.builder(httpHosts.toArray(new HttpHost[0]))); } else { client = new RestHighLevelClient( RestClient.builder(new HttpHost("127.0.0.1", 9200, "http"))); } return client; } private IndexRequest getIndexRequest(String index, String indexType, String docId, Map<String, Object> dataMap) { IndexRequest indexRequest = null; if (null == index || null == indexType) { throw new ElasticsearchException("index or indexType must not be null"); } if (null == docId) { indexRequest = new IndexRequest(index, indexType); } else { indexRequest = new IndexRequest(index, indexType, docId); } return indexRequest; } /** * 同步執行索引 * * @param index * @param indexType * @param docId * @param dataMap * @throws IOException */ public IndexResponse execIndex(String index, String indexType, String docId, Map<String, Object> dataMap) throws IOException { return getClient().index(getIndexRequest(index, indexType, docId, dataMap).source(dataMap)); } /** * 異步執行 * * @param index * @param indexType * @param docId * @param dataMap * @param indexResponseActionListener * @throws IOException */ public void asyncExecIndex(String index, String indexType, String docId, Map<String, Object> dataMap, ActionListener<IndexResponse> indexResponseActionListener) throws IOException { getClient().indexAsync(getIndexRequest(index, indexType, docId, dataMap).source(dataMap), indexResponseActionListener); } /** * @param index * @param indexType * @param docId * @param includes 返回需要包含的字段，可以傳入空 * @param excludes 返回需要不包含的字段，可以傳入為空 * @param excludes version * @param excludes versionType * @return * @throws IOException */ public GetResponse getRequest(String index, String indexType, String docId, String[] includes, String[] excludes, Integer version, VersionType versionType) throws IOException { if (null == includes || includes.length == 0) { includes = Strings.EMPTY_ARRAY; } if (null == excludes || excludes.length == 0) { excludes = Strings.EMPTY_ARRAY; } GetRequest getRequest = new GetRequest(index, indexType, docId); FetchSourceContext fetchSourceContext = new FetchSourceContext(true, includes, excludes); getRequest.realtime(true); if (null != version) { getRequest.version(version); } if (null != versionType) { getRequest.versionType(versionType); } return getClient().get(getRequest.fetchSourceContext(fetchSourceContext)); } /** * @param index * @param indexType * @param docId * @param includes * @param excludes * @return * @throws IOException */ public GetResponse getRequest(String index, String indexType, String docId, String[] includes, String[] excludes) throws IOException { return getRequest(index, indexType, docId, includes, excludes, null, null); } /** * @param index * @param indexType * @param docId * @return * @throws IOException */ public GetResponse getRequest(String index, String indexType, String docId) throws IOException { GetRequest getRequest = new GetRequest(index, indexType, docId); return getClient().get(getRequest); } /** * @param index * @param indexType * @param docId * @return * @throws IOException */ public Boolean existDoc(String index, String indexType, String docId) throws IOException { GetRequest getRequest = new GetRequest(index, indexType, docId); getRequest.fetchSourceContext(new FetchSourceContext(false)); getRequest.storedFields("_none_"); return getClient().exists(getRequest); } /** * @param index * @param indexType * @param docId * @param timeValue * @param refreshPolicy * @param version * @param versionType * @return * @throws IOException */ public DeleteResponse deleteDoc(String index, String indexType, String docId, TimeValue timeValue, WriteRequest.RefreshPolicy refreshPolicy, Integer version, VersionType versionType) throws IOException { DeleteRequest deleteRequest = new DeleteRequest(index, indexType, docId); if (null != timeValue) { deleteRequest.timeout(timeValue); } if (null != refreshPolicy) { deleteRequest.setRefreshPolicy(refreshPolicy); } if (null != version) { deleteRequest.version(version); } if (null != versionType) { deleteRequest.versionType(versionType); } return getClient().delete(deleteRequest); } /** * @param index * @param indexType * @param docId * @return * @throws IOException */ public DeleteResponse deleteDoc(String index, String indexType, String docId) throws IOException { return deleteDoc(index, indexType, docId, null, null, null, null); } /** * @param index * @param indexType * @param docId * @param dataMap * @param timeValue * @param refreshPolicy * @param version * @param versionType * @param docAsUpsert * @param includes * @param excludes * @return * @throws IOException */ public UpdateResponse updateDoc(String index, String indexType, String docId, Map<String, Object> dataMap, TimeValue timeValue, WriteRequest.RefreshPolicy refreshPolicy, Integer version, VersionType versionType, Boolean docAsUpsert, String[] includes, String[] excludes) throws IOException { UpdateRequest updateRequest = new UpdateRequest(index, indexType, docId); updateRequest.doc(dataMap); if (null != timeValue) { updateRequest.timeout(timeValue); } if (null != refreshPolicy) { updateRequest.setRefreshPolicy(refreshPolicy); } if (null != version) { updateRequest.version(version); } if (null != versionType) { updateRequest.versionType(versionType); } updateRequest.docAsUpsert(docAsUpsert); //衝突時重試的次數 updateRequest.retryOnConflict(3); if (null == includes && null == excludes) { return getClient().update(updateRequest); } else { if (null == includes || includes.length == 0) { includes = Strings.EMPTY_ARRAY; } if (null == excludes || excludes.length == 0) { excludes = Strings.EMPTY_ARRAY; } return getClient().update(updateRequest.fetchSource(new FetchSourceContext(true, includes, excludes))); } } /** * 更新時不存在就插入 * * @param index * @param indexType * @param docId * @param dataMap * @return * @throws IOException */ public UpdateResponse upDdateocAsUpsert(String index, String indexType, String docId, Map<String, Object> dataMap) throws IOException { return updateDoc(index, indexType, docId, dataMap, null, null, null, null, true, null, null); } /** * 存在才更新 * * @param index * @param indexType * @param docId * @param dataMap * @return * @throws IOException */ public UpdateResponse updateDoc(String index, String indexType, String docId, Map<String, Object> dataMap) throws IOException { return updateDoc(index, indexType, docId, dataMap, null, null, null, null, false, null, null); } /** * 批量操作 * * @param indexBeanList * @param timeValue * @param refreshPolicy * @return * @throws IOException */ public BulkResponse bulkRequest(List<IndexBean> indexBeanList, TimeValue timeValue, WriteRequest.RefreshPolicy refreshPolicy) throws IOException { BulkRequest bulkRequest = getBulkRequest(indexBeanList); if (null != timeValue) { bulkRequest.timeout(timeValue); } if (null != refreshPolicy) { bulkRequest.setRefreshPolicy(refreshPolicy); } return getClient().bulk(bulkRequest); } private BulkRequest getBulkRequest(List<IndexBean> indexBeanList) { BulkRequest bulkRequest = new BulkRequest(); indexBeanList.forEach(indexBean -> { if ("1".equals(indexBean.getOperateType())) { bulkRequest.add(null != indexBean.getDocId() ? new IndexRequest(indexBean.getIndex(), indexBean.getIndexType(), indexBean.getDocId()) : new IndexRequest(indexBean.getIndex(), indexBean.getIndexType())); } else if ("2".equals(indexBean.getOperateType())) { if ((null != indexBean.getDocId())) { throw new ElasticsearchException("update action docId must not be null"); } bulkRequest.add(new UpdateRequest(indexBean.getIndex(), indexBean.getIndexType(), indexBean.getDocId())); } else if ("3".equals(indexBean.getOperateType())) { if ((null != indexBean.getDocId())) { throw new ElasticsearchException("delete action docId must not be null"); } bulkRequest.add(new DeleteRequest(indexBean.getIndex(), indexBean.getIndexType(), indexBean.getDocId())); } else { throw new ElasticsearchException("OperateType" + indexBean.getOperateType() + "is not support"); } }); return bulkRequest; } /** * 批量操作 * * @param indexBeanList * @return */ public BulkResponse bulkRequest(List<IndexBean> indexBeanList) throws IOException { return bulkRequest(indexBeanList, null, null); } /** * 批量異步操作 * * @param indexBeanList * @param bulkResponseActionListener */ public void AsyncBulkRequest(List<IndexBean> indexBeanList, ActionListener<BulkResponse> bulkResponseActionListener) { getClient().bulkAsync(getBulkRequest(indexBeanList), bulkResponseActionListener); } private SearchRequest getSearchRequest(String index, String indexType) { SearchRequest searchRequest; if (null == index) { throw new ElasticsearchException("index name must not be null"); } if (null != indexType) { searchRequest = new SearchRequest(index, indexType); } else { searchRequest = new SearchRequest(index); } return searchRequest; } /** * @param index * @param indexType * @return * @throws IOException */ public SearchResponse searchRequest(String index, String indexType) throws IOException { return getClient().search(getSearchRequest(index, indexType)); } /** * @param index * @param indexType * @param from * @param size * @param termQueryBuilder * @return * @throws IOException */ public SearchResponse searchRequest(String index, String indexType, Integer from, Integer size, TermQueryBuilder termQueryBuilder) throws IOException { return getClient().search(getSearchRequest(index, indexType).source(getSearchSourceBuilder(index, indexType, from, size, termQueryBuilder, null, null, null))); } private SearchSourceBuilder getSearchSourceBuilder(String index, String indexType, Integer from, Integer size, TermQueryBuilder termQueryBuilder, String sortField, SortBuilder sortBuilder, Boolean fetchSource) { SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder(); if (null != termQueryBuilder) { searchSourceBuilder.query(termQueryBuilder); } searchSourceBuilder.from(from); searchSourceBuilder.size(size); if (null != sortField) { searchSourceBuilder.sort(sortField); } if (null != sortBuilder) { searchSourceBuilder.sort(sortBuilder); } //設置超時時間 searchSourceBuilder.timeout(new TimeValue(120, TimeUnit.SECONDS)); if (null != fetchSource) { searchSourceBuilder.fetchSource(fetchSource); } return searchSourceBuilder; } /** * @param index * @param indexType * @param from * @param size * @param termQueryBuilder * @param matchQueryBuilder * @return * @throws IOException */ public SearchResponse searchRequest(String index, String indexType, Integer from, Integer size, TermQueryBuilder termQueryBuilder, MatchQueryBuilder matchQueryBuilder) throws IOException { if (null == matchQueryBuilder) { throw new ElasticsearchException("matchQueryBuilder is null"); } return getClient().search(getSearchRequest(index, indexType).source(getSearchSourceBuilder(index, indexType, from, size, termQueryBuilder, null, null, null).query(matchQueryBuilder))); } /** * @param index * @param indexType * @param from * @param size * @param matchQueryBuilder * @return * @throws IOException */ public SearchResponse searchRequest(String index, String indexType, Integer from, Integer size, MatchQueryBuilder matchQueryBuilder) throws IOException { if (null == matchQueryBuilder) { throw new ElasticsearchException("matchQueryBuilder is null"); } return getClient().search(getSearchRequest(index, indexType).source(getSearchSourceBuilder(index, indexType, from, size, null, null, null, null).query(matchQueryBuilder))); } /** * @param index * @param indexType * @param from * @param size * @param matchQueryBuilder * @param sortField * @return * @throws IOException */ public SearchResponse searchRequest(String index, String indexType, Integer from, Integer size, MatchQueryBuilder matchQueryBuilder, String sortField) throws IOException { if (null == matchQueryBuilder) { throw new ElasticsearchException("matchQueryBuilder is null"); } return getClient().search(getSearchRequest(index, indexType).source(getSearchSourceBuilder(index, indexType, from, size, null, sortField, null, null).query(matchQueryBuilder))); } /** * @param index * @param indexType * @param from * @param size * @param matchQueryBuilder * @param sortField * @param fetchSource * @return * @throws IOException */ public SearchResponse searchRequest(String index, String indexType, Integer from, Integer size, MatchQueryBuilder matchQueryBuilder, String sortField, Boolean fetchSource) throws IOException { if (null == matchQueryBuilder) { throw new ElasticsearchException("matchQueryBuilder is null"); } return getClient().search(getSearchRequest(index, indexType).source(getSearchSourceBuilder(index, indexType, from, size, null, sortField, null, fetchSource).query(matchQueryBuilder))); } /** * @param index * @param indexType * @param from * @param size * @param matchQueryBuilder * @param sortBuilder * @return * @throws IOException */ public SearchResponse searchRequest(String index, String indexType, Integer from, Integer size, MatchQueryBuilder matchQueryBuilder, SortBuilder sortBuilder) throws IOException { if (null == matchQueryBuilder) { throw new ElasticsearchException("matchQueryBuilder is null"); } return getClient().search(getSearchRequest(index, indexType).source(getSearchSourceBuilder(index, indexType, from, size, null, null, sortBuilder, null).query(matchQueryBuilder))); } /** * 支持排序 * * @param index * @param indexType * @param from * @param size * @param termQueryBuilder * @param matchQueryBuilder * @param sortField * @return * @throws IOException */ public SearchResponse searchRequest(String index, String indexType, Integer from, Integer size, TermQueryBuilder termQueryBuilder, MatchQueryBuilder matchQueryBuilder, String sortField) throws IOException { if (null == matchQueryBuilder) { throw new ElasticsearchException("matchQueryBuilder is null"); } return getClient().search(getSearchRequest(index, indexType).source(getSearchSourceBuilder(index, indexType, from, size, termQueryBuilder, sortField, null, null).query(matchQueryBuilder))); } /** * @param index * @param indexType * @param from * @param size * @param termQueryBuilder * @param matchQueryBuilder * @param sortBuilder * @param fetchSource 開關 * @return * @throws IOException */ public SearchResponse searchRequest(String index, String indexType, Integer from, Integer size, TermQueryBuilder termQueryBuilder, MatchQueryBuilder matchQueryBuilder, SortBuilder sortBuilder, Boolean fetchSource) throws IOException { if (null == matchQueryBuilder) { throw new ElasticsearchException("matchQueryBuilder is null"); } return getClient().search(getSearchRequest(index, indexType).source(getSearchSourceBuilder(index, indexType, from, size, termQueryBuilder, null, sortBuilder, fetchSource).query(matchQueryBuilder))); } /** * @param index * @param indexType * @param from * @param size * @param termQueryBuilder * @param matchQueryBuilder * @param sortBuilder * @return * @throws IOException */ public SearchResponse searchRequest(String index, String indexType, Integer from, Integer size, TermQueryBuilder termQueryBuilder, MatchQueryBuilder matchQueryBuilder, SortBuilder sortBuilder) throws IOException { if (null == matchQueryBuilder) { throw new ElasticsearchException("matchQueryBuilder is null"); } return getClient().search(getSearchRequest(index, indexType).source(getSearchSourceBuilder(index, indexType, from, size, termQueryBuilder, null, sortBuilder, null).query(matchQueryBuilder))); } /** * @param index * @param indexType * @param from * @param size * @param termQueryBuilder * @param matchQueryBuilder * @param sortField * @param fetchSource * @return * @throws IOException */ public SearchResponse searchRequest(String index, String indexType, Integer from, Integer size, TermQueryBuilder termQueryBuilder, MatchQueryBuilder matchQueryBuilder, String sortField, Boolean fetchSource) throws IOException { if (null == matchQueryBuilder) { throw new ElasticsearchException("matchQueryBuilder is null"); } return getClient().search(getSearchRequest(index, indexType).source(getSearchSourceBuilder(index, indexType, from, size, termQueryBuilder, sortField, null, fetchSource).query(matchQueryBuilder))); } /** * 異步操作 * @param index * @param indexType * @param from * @param size * @param termQueryBuilder * @param matchQueryBuilder * @param sortBuilder * @param listener * @throws IOException */ public void asyncSearchRequest(String index, String indexType, Integer from, Integer size, TermQueryBuilder termQueryBuilder, MatchQueryBuilder matchQueryBuilder, SortBuilder sortBuilder,ActionListener<SearchResponse> listener) throws IOException { if (null == matchQueryBuilder) { throw new ElasticsearchException("matchQueryBuilder is null"); } getClient().searchAsync(getSearchRequest(index, indexType).source(getSearchSourceBuilder(index, indexType, from, size, termQueryBuilder, null, sortBuilder, null).query(matchQueryBuilder)),listener); } /** * 異步操作 * @param index * @param indexType * @param from * @param size * @param termQueryBuilder * @param matchQueryBuilder * @param sortField * @param listener * @throws IOException */ public void asyncSearchRequest(String index, String indexType, Integer from, Integer size, TermQueryBuilder termQueryBuilder, MatchQueryBuilder matchQueryBuilder, String sortField,ActionListener<SearchResponse> listener) throws IOException { if (null == matchQueryBuilder) { throw new ElasticsearchException("matchQueryBuilder is null"); } getClient().searchAsync(getSearchRequest(index, indexType).source(getSearchSourceBuilder(index, indexType, from, size, termQueryBuilder, sortField, null, null).query(matchQueryBuilder)),listener); } }
二、transportClient API
未完待續，近期繼續整理
