Elasticsearch Java API 很全的整理以及架構剖析

Elasticsearch 的API 分為 REST Client API（http請求形式）以及 transportClient API兩種，相比來說transportClient API效率更高，transportClient 是通過Elasticsearch內部RPC的形式進行請求的，連接可以是一個長連接，相當于是把客戶端的請求當成

Elasticsearch 集群的一個節點，當然 REST Client API 也支持http keepAlive形式的長連接，只是非內部RPC形式，但是從Elasticsearch 7 后就會移除transportClient ，主要原因是transportClient 難以向下兼容版本，

本文中所有的講解和操作都是基于jdk 1.8 和elasticsearch 6.2.4版本，

備注：本文參考了很多Elasticsearch 的官方檔案以及部l網路資料做的綜合整理，  本人github 參考代碼：https://github.com/597365581/bigdata_tools/tree/master/yongqing-bigdata-tools/yongqing-elasticsearch-tool

一、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 可以是下面的格式

本文作者：張永清，轉載請注明出處：Elasticsearch Java API 很全的整理

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) {
        // 版本沖突
    }
}

封裝好的參考方法：

本文作者：張永清，轉載請注明出處：Elasticsearch Java API 很全的整理

  /**
     * @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 串列

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()

9、關于 REST Client的完整工具類代碼

本文作者：張永清，轉載請注明出處：Elasticsearch Java API 很全的整理

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

未完待續，近期繼續整理

三、Elasticsearch 架構

1、基礎概念：

1）、集群（Cluster): 包含一個或多個具有相同 cluster.name 的節點.

• 集群內節點協同作業，共享資料，并共同分擔作業負荷，
• 由于節點是從屬集群的，集群會自我重組來均勻地分發資料. 
• cluster Name是很重要的，因為每個節點只能是群集的一部分，當該節點被設定為相同的名稱時，就會自動加入群集，
• 集群中通過選舉產生一個mater節點，它將負責管理集群范疇的變更，例如創建或洗掉索引，添加節點到集群或從集群洗掉節點，master 節點無需參與檔案層面的變更和搜索，這意味著僅有一個 master 節點并不會因流量增長而成為瓶頸，任意一個節點都可以成為 master 節點，我們例舉的集群只有一個節點，因此它會扮演 master 節點的角色，
• 作為用戶，我們可以訪問包括 master 節點在內的集群中的任一節點，每個節點都知道各個檔案的位置，并能夠將我們的請求直接轉發到擁有我們想要的資料的節點，無論我們訪問的是哪個節點，它都會控制從擁有資料的節點收集回應的程序，并回傳給客戶端最終的結果，這一切都是由 Elasticsearch 透明管理的

2）、節點(node):  一個節點是一個邏輯上獨立的服務，可以存盤資料，并參與集群的索引和搜索功能, 一個節點也有唯一的名字，群集通過節點名稱進行管理和通信.

3）、索引（Index)： 索引與關系型資料庫實體(Database)相當，索引只是一個 邏輯命名空間，它指向一個或多個分片(shards)，內部用Apache Lucene實作索引中資料的讀寫

4）、檔案型別（Type）：相當于資料庫中的table概念，每個檔案在ElasticSearch中都必須設定它的型別，檔案型別使得同一個索引中在存盤結構不同檔案時，只需要依據檔案型別就可以找到對應的引數映射(Mapping)資訊，方便檔案的存取

5）、檔案（Document) ：相當于資料庫中的row， 是可以被索引的基本單位，例如，你可以有一個的客戶檔案，有一個產品檔案，還有一個訂單的檔案，檔案是以JSON格式存盤的，在一個索引中，您可以存盤多個的檔案，請注意，雖然在一個索引中有多分檔案，但這些檔案的結構是一致的，并在第一次存盤的時候指定, 檔案屬于一種 型別(type)，各種各樣的型別存在于一個 索引 中，你也可以通過類比傳統的關系資料庫得到一些大致的相似之處：

6）、Mapping： 相當于資料庫中的schema，用來約束欄位的型別，不過 Elasticsearch 的 mapping 可以自動根據資料創建

7）、分片(shard) ：是 作業單元(worker unit) 底層的一員，用來分配集群中的資料，它只負責保存索引中所有資料的一小片，

• 分片是一個獨立的Lucene實體，并且它自身也是一個完整的搜索引擎，
• 檔案存盤并且被索引在分片中，但是我們的程式并不會直接與它們通信，取而代之，它們直接與索引進行通信的
• 把分片想象成一個資料的容器，資料被存盤在分片中，然后分片又被分配在集群的節點上，當你的集群擴展或者縮小時，elasticsearch 會自動的在節點之間遷移分配分片，以便集群保持均衡
• 分片分為 主分片(primary shard) 以及 從分片(replica shard) 兩種，在你的索引中，每一個檔案都屬于一個主分片
• 從分片只是主分片的一個副本，它用于提供資料的冗余副本，在硬體故障時提供資料保護，同時服務于搜索和檢索這種只讀請求
• 索引中的主分片的數量在索引創建后就固定下來了，但是從分片的數量可以隨時改變，
• 一個索引默認設定了5個主分片，每個主分片有一個從分片對應

2、ES模塊

 

 

1）、 Gateway: 代表ES的持久化存盤方式，包含索引資訊，ClusterState(集群資訊)，mapping，索引碎片資訊，以及transaction log等

• 對于分布式集群來說，當一個或多個節點down掉了，能夠保證我們的資料不能丟，最通用的解放方案就是對失敗節點的資料進行復制，通過控制復制的份數可以保證集群有很高的可用性，復制這個方案的精髓主要是保證操作的時候沒有單點，對一個節點的操作會同步到其他的復制節點上去，
• ES一個索引會拆分成多個碎片，每個碎片可以擁有一個或多個副本（創建索引的時候可以配置），這里有個例子，每個索引分成3個碎片，每個碎片有2個副本，如下：

$ curl -XPUT http://localhost:9200/twitter/ -d '
index :
    number_of_shards : 3
    number_of_replicas : 2

• 每個操作會自動路由主碎片所在的節點，在上面執行操作，并且同步到其他復制節點，通過使用“non blocking IO”模式所有復制的操作都是并行執行的，也就是說如果你的節點的副本越多，你網路上的流量消耗也會越大，復制節點同樣接受來自外面的讀操作，意義就是你的復制節點越多，你的索引的可用性就越強，對搜索的可伸縮行就更好，能夠承載更多的操作
• 第一次啟動的時候，它會去持久化設備讀取集群的狀態資訊（創建的索引，配置等）然后執行應用它們（創建索引，創建mapping映射等），每一次shard節點第一次實體化加入復制組，它都會從長持久化存盤里面恢復它的狀態資訊

2）、 Lucence Directory:  是lucene的框架服務發現以及選主 ZenDiscovery: 用來實作節點自動發現，還有Master節點選取，假如Master出現故障，其它的這個節點會自動選舉，產生一個新的Master

它是Lucene存盤的一個抽象，由此派生了兩個類：FSDirectory和RAMDirectory，用于控制索引檔案的存盤位置，使用FSDirectory類，就是存盤到硬碟；使用RAMDirectory類，則是存盤到記憶體

 

 一個Directory物件是一份檔案的清單，檔案可能只在被創建的時候寫一次，一旦檔案被創建，它將只被讀取或者洗掉，在讀取的時候進行寫入操作是允許的

3）、Discovery

• 節點啟動后先ping（這里的ping是 Elasticsearch 的一個RPC命令，如果 discovery.zen.ping.unicast.hosts 有設定，則ping設定中的host，否則嘗試ping localhost 的幾個埠， Elasticsearch 支持同一個主機啟動多個節點）
• Ping的response會包含該節點的基本資訊以及該節點認為的master節點
• 選舉開始，先從各節點認為的master中選，規則很簡單，按照id的字典序排序，取第一個
• 如果各節點都沒有認為的master，則從所有節點中選擇，規則同上，這里有個限制條件就是 discovery.zen.minimum_master_nodes，如果節點數達不到最小值的限制，則回圈上述程序，直到節點數足夠可以開始選舉
• 最后選舉結果是肯定能選舉出一個master，如果只有一個local節點那就選出的是自己
• 如果當前節點是master，則開始等待節點數達到 minimum_master_nodes，然后提供服務, 如果當前節點不是master，則嘗試加入master.
• ES支持任意數目的集群（1-N）,所以不能像 Zookeeper/Etcd 那樣限制節點必須是奇數，也就無法用投票的機制來選主，而是通過一個規則，只要所有的節點都遵循同樣的規則，得到的資訊都是對等的，選出來的主節點肯定是一致的. 但分布式系統的問題就出在資訊不對等的情況，這時候很容易出現腦裂（Split-Brain）的問題，大多數解決方案就是設定一個quorum值，要求可用節點必須大于quorum（一般是超過半數節點），才能對外提供服務，而 Elasticsearch 中，這個quorum的配置就是 discovery.zen.minimum_master_nodes ，

4）、memcached

• 通過memecached協議來訪問ES的介面,支持二進制和文本兩種協議.通過一個名為transport-memcached插件提供
• Memcached命令會被映射到REST介面，并且會被同樣的REST層處理，memcached命令串列包括：get/set/delete/quit

5）、River ：代表es的一個資料源，也是其它存盤方式（如：資料庫）同步資料到es的一個方法，它是以插件方式存在的一個es服務，通過讀取river中的資料并把它索引到es中，官方的river有couchDB的，RabbitMQ的，Twitter的，Wikipedia的，

本文作者：張永清，轉載請注明出處：Elasticsearch Java API 很全的整理  最近發現很多轉載了本博客的文章不注明出處，作者將追究法律責任

 

轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/37954.html

標籤：NoSQL

上一篇：一文了解 Redis 記憶體監控和記憶體消耗

下一篇：MongoDB 備份與恢復