問題

一次在使用Elasticsearch來存儲(chǔ)數(shù)據(jù)的時(shí)候，前端在把數(shù)據(jù)保存成功后，調(diào)用查詢列表的接口來刷新列表時(shí)，發(fā)現(xiàn)有時(shí)能刷新出新保存的數(shù)據(jù)，有時(shí)候又不能，修改和刪除數(shù)據(jù)也發(fā)現(xiàn)有類似現(xiàn)象。但是再次刷新頁面的時(shí)候又可以查詢到新操作的數(shù)據(jù)。

后面發(fā)現(xiàn)這個(gè)問題是Elasticsearch本身在操作數(shù)據(jù)后有一定的延遲性導(dǎo)致。Elasticsearch默認(rèn)情況下在寫入數(shù)據(jù)后，是要等1s后才能被查詢到。

因?yàn)镋lasticsearch中每次索引refresh會(huì)產(chǎn)生一個(gè)新的 lucene 段，這會(huì)導(dǎo)致頻繁的 segment merge 行為，對系統(tǒng) CPU 和 IO 占用都比較高。

解決方法1

需要修改ElasticSearch的刷新策略

org.elasticsearch.action.support.WriteRequest.RefreshPolicy 中定義了三種刷新策略：

enum RefreshPolicy implements Writeable {
    /**
     * Don't refresh after this request. The default.
     */
    NONE("false"),
    /**
     * Force a refresh as part of this request. This refresh policy does not scale for high indexing or search throughput but is useful
     * to present a consistent view to for indices with very low traffic. And it is wonderful for tests!
     */
    IMMEDIATE("true"),
    /**
     * Leave this request open until a refresh has made the contents of this request visible to search. This refresh policy is
     * compatible with high indexing and search throughput but it causes the request to wait to reply until a refresh occurs.
     */
    WAIT_UNTIL("wait_for");

IMMEDIATE

• 請求向ElasticSearch提交了數(shù)據(jù)，立即刷新數(shù)據(jù)后，再結(jié)束請求。
• 優(yōu)點(diǎn)：操作延時(shí)短、實(shí)時(shí)性高
• 缺點(diǎn)：資源消耗高

WAIT_UNTIL

• 請求向ElasticSearch提交了數(shù)據(jù)，等待數(shù)據(jù)完成刷新（默認(rèn)為1s），然后再結(jié)束請求。
• 優(yōu)點(diǎn)：實(shí)時(shí)性高、操作延時(shí)長
• 缺點(diǎn)：資源消耗低

NONE (默認(rèn)策略)

• 請求向ElasticSearch提交了數(shù)據(jù)，不等待數(shù)據(jù)完成刷新，直接結(jié)束請求。
• 優(yōu)點(diǎn)：操作延時(shí)短、資源消耗低
• 缺點(diǎn)：實(shí)時(shí)性低

org.elasticsearch.action.support.WriteRequestBuilder中的setRefreshPolicy方法，默認(rèn)為設(shè)置為RefreshPolicy#NONE

/**
 * Should this request trigger a refresh ({@linkplain RefreshPolicy#IMMEDIATE}), wait for a refresh (
 * {@linkplain RefreshPolicy#WAIT_UNTIL}), or proceed ignore refreshes entirely ({@linkplain RefreshPolicy#NONE}, the default).
 */
@SuppressWarnings("unchecked")
default B setRefreshPolicy(RefreshPolicy refreshPolicy) {
    request().setRefreshPolicy(refreshPolicy);
    return (B) this;
}

代碼中設(shè)置 setRefreshPolicy

setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE) 設(shè)置為立即刷新，即可解決不能立即查詢到數(shù)據(jù)的問題。

// 刪除 設(shè)置為WriteRequest.RefreshPolicy.IMMEDIATE 立即刷新
DeleteRequestBuilder deleteRequestBuilder = transportClient.prepareDelete(INDEX, TYPE, id).setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE); 
// 新增
IndexRequestBuilder indexRequestBuilder = transportClient.prepareIndex(INDEX, TYPE, id).setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
// 更新
UpdateRequestBuilder updateRequestBuilder = transportClient.prepareUpdate(INDEX, TYPE,id).setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
// 批量
BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk().setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);

支持的接口：

• 刪除：DeleteRequestBuilder
• 新增：IndexRequestBuilder
• 更新：UpdateRequestBuilder
• 批量：BulkRequestBuilder

解決方法2

在查詢結(jié)果前使用sleep來暫停一段時(shí)間再執(zhí)行查詢操作，這樣可以不改變ElasticSearch的刷新策略。

當(dāng)業(yè)務(wù)需要頻繁的操作修改數(shù)據(jù)的情況下，設(shè)置刷新策略為立即刷新的話可能會(huì)造成性能問題。

我們可以使用sleep的方式不去修改默認(rèn)的刷新策略。

當(dāng)我們在執(zhí)行新增、修改、刪除、批量操作的時(shí)候，在這些操作中設(shè)置一把鎖，過期時(shí)間為1s（這個(gè)時(shí)間剛好是刷新策略的默認(rèn)時(shí)間），在查詢數(shù)據(jù)時(shí)我們需要先去檢查是否有鎖，如果有鎖存在我們就等待500ms后再執(zhí)行查詢操作。這樣基本上又可以保證不改刷新策略，也可以保存查詢的實(shí)時(shí)性。

/**
 * 在新增、修改、刪除、批量操作的時(shí)候設(shè)置鎖
 */
private void setLock() {
  redisService.setWithExpire(getEsParentUpdateKey(), UPDATING_MSG, 1);
}

/**
 * 查詢r(jià)edis中的鎖是否存在，如果有鎖，將暫停500毫秒，再執(zhí)行查詢操作
 */
private void waitMe() {
    Object lock = redisService.get(getEsParentUpdateKey());
    if (lock != null) {
        try {
            //為了得到ES更新后的數(shù)據(jù)，暫停500s
            TimeUnit.MICROSECONDS.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}