Elasticsearch Java API 很全的整理

時間 2019-11-07

標籤 html java 正則表達式算法 shell apache 編程 json 網絡併發欄目日誌分析简体版

原文原文鏈接

Elasticsearch 的API 分爲 REST Client API（http請求形式）以及 transportClient API兩種。相比來講transportClient API效率更高，transportClient 是經過Elasticsearch內部RPC的形式進行請求的，鏈接能夠是一個長鏈接，至關因而把客戶端的請求當成html

Elasticsearch 集羣的一個節點。可是從Elasticsearch 7 後就會移除transportClient 。主要緣由是transportClient 難以向下兼容版本。java

本文中全部的講解和操做都是基於jdk 1.8 和elasticsearch 6.2.4版本。正則表達式

備註：本文參考了不少Elasticsearch 的官方文檔以及部l網絡資料作的綜合整理。算法

1、High REST Clientshell

High Client 基於 Low Client，主要目的是暴露一些 API，這些 API 能夠接受請求對象爲參數，返回響應對象，而對請求和響應細節的處理都是由 client 自動完成的。apache

API 在調用時均可以是同步或者異步兩種形式
同步 API 會致使阻塞，一直等待數據返回
異步 API 在命名上會加上 async 後綴，須要有一個 listener 做爲參數，等這個請求返回結果或者發生錯誤時，這個 listener 就會被調用，listener主要是解決自動回調的問題，有點像安卓開發裏面的listener監聽回調。編程

Elasticsearch REST APi 官方地址：https://www.elastic.co/guide/en/elasticsearch/reference/6.2/index.htmljson

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

一、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) {
        
    }
}

二、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);
    }

三、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);
    }

四、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);
    }

五、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);
    }

六、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);
    }

七、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 之間都是嵌套關係，能夠參考下圖：

八、全文查詢 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 列表

Search Query	QueryBuilder Class	Method in QueryBuilders
Match	MatchQueryBuilder	QueryBuilders.matchQuery()
Match Phrase	MatchPhraseQueryBuilder	QueryBuilders.matchPhraseQuery()
Match Phrase Prefix	MatchPhrasePrefixQueryBuilder	QueryBuilders.matchPhrasePrefixQuery()
Multi Match	MultiMatchQueryBuilder	QueryBuilders.multiMatchQuery()
Common Terms	CommonTermsQueryBuilder	QueryBuilders.commonTermsQuery()
Query String	QueryStringQueryBuilder	QueryBuilders.queryStringQuery()
Simple Query String	SimpleQueryStringBuilder	QueryBuilders.simpleQueryStringQuery()

基於詞項的查詢
這種類型的查詢不須要分析，它們是對單個詞項操做，只是在倒排索引中查找準確的詞項（精確匹配）而且使用 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 列表

Search Query	QueryBuilder Class	Method in QueryBuilders
Term	TermQueryBuilder	QueryBuilders.termQuery()
Terms	TermsQueryBuilder	QueryBuilders.termsQuery()
Range	RangeQueryBuilder	QueryBuilders.rangeQuery()
Exists	ExistsQueryBuilder	QueryBuilders.existsQuery()
Prefix	PrefixQueryBuilder	QueryBuilders.prefixQuery()
Wildcard	WildcardQueryBuilder	QueryBuilders.wildcardQuery()
Regexp	RegexpQueryBuilder	QueryBuilders.regexpQuery()
Fuzzy	FuzzyQueryBuilder	QueryBuilders.fuzzyQuery()
Type	TypeQueryBuilder	QueryBuilders.typeQuery()
Ids	IdsQueryBuilder	QueryBuilders.idsQuery()

複合查詢
什麼是複合查詢？
複合查詢會將其餘的複合查詢或者葉查詢包裹起來，以嵌套的形式展現和執行，獲得的結果也是對各個子查詢結果和分數的合併。能夠分爲下面幾種：

constant_score query

常常用在使用 filter 的場合，全部匹配的文檔分數都是一個不變的常量

bool query

能夠將多個葉查詢和組合查詢再組合起來，可接受的參數以下

must : 文檔必須匹配這些條件才能被包含進來
must_not 文檔必須不匹配才能被包含進來
should 若是知足其中的任何語句，都會增長分數；即便不知足，也沒有影響
filter 以過濾模式進行，不評分，可是必須匹配
dis_max query

叫作分離最大化查詢，它會將任何與查詢匹配的文檔都做爲結果返回，可是隻是將其中最佳匹配的評分做爲最終的評分返回。

function_score query

容許爲每一個與主查詢匹配的文檔應用一個函數，可用來改變甚至替換原始的評分

boosting query

用來控制（提升或下降）複合查詢中子查詢的權重。

Search Query	QueryBuilder Class	Method in QueryBuilders
Constant Score	ConstantScoreQueryBuilder	QueryBuilders.constantScoreQuery()
Bool	BoolQueryBuilder	QueryBuilders.boolQuery()
Dis Max	DisMaxQueryBuilder	QueryBuilders.disMaxQuery()
Function Score	FunctionScoreQueryBuilder	QueryBuilders.functionScoreQuery()
Boosting	BoostingQueryBuilder	QueryBuilders.boostingQuery()

特殊查詢
Wrapper Query
這裏比較重要的一個是 Wrapper Query，是說能夠接受任何其餘 base64 編碼的字符串做爲子查詢。

主要應用場合就是在 Rest High-Level REST client 中接受 json 字符串做爲參數。好比使用 gson 等 json 庫將要查詢的語句拼接好，直接塞到 Wrapper Query 中查詢就能夠了，很是方便。

Wrapper Query 對應的 QueryBuilder class 是WrapperQueryBuilder

具體方法是 QueryBuilders.wrapperQuery()

九、關於 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);
    }
}