一、前言

介紹我們?cè)谇懊嬉呀?jīng)知道ElasticSearch底層的寫(xiě)入是基于lucence依進(jìn)行doc寫(xiě)入的。ElasticSearch作為一款分布式系統(tǒng)，在寫(xiě)入數(shù)據(jù)時(shí)還需要考慮很多重要的事項(xiàng)，比如:可靠性、原子性、一致性、實(shí)時(shí)性、隔離性、性能等多個(gè)指標(biāo)。

ElasticSearch是如何做到的呢？下面我們針對(duì)ElasticSearch的寫(xiě)入進(jìn)行分析。

二、lucence寫(xiě)

2.1 增刪改

ElasticSearch拿到一個(gè)doc后調(diào)用lucence的api進(jìn)行寫(xiě)入的。

 public long addDocument();
 public long updateDocuments();
 public long deleteDocuments();

如上面的代碼所示，我們使用lucence的上面的接口就可以完成文檔的增刪改操作。在lucence中有一個(gè)核心的類IndexWriter負(fù)責(zé)數(shù)據(jù)寫(xiě)入和索引相關(guān)的工作。

//1. 初始化indexwriter對(duì)象
IndexWriter writer = new IndexWriter(new Directory(Paths.get("/index")), new IndexWriterConfig());
//2. 創(chuàng)建文檔
Document doc = new Document();
doc.add(new StringField("empName", "王某某", Field.Store.YES));
doc.add(new TextField("content", "操作了某菜單", Field.Store.YES));
//3. 添加文檔
writer.addDocument(doc);
//4. 提交
writer.commit();

以上代碼演示了最基礎(chǔ)的lucence的寫(xiě)入操作，主要涉及到幾個(gè)關(guān)鍵點(diǎn)： 初始化： Directory是負(fù)責(zé)持久化的，他的具體實(shí)現(xiàn)有很多，有本地文件系統(tǒng)、數(shù)據(jù)庫(kù)、分布式文件系統(tǒng)等待，ElasticSearch默認(rèn)的實(shí)現(xiàn)是本地文件系統(tǒng)。 Document： Document就是es中的文檔，F(xiàn)iledType定義了很多索引類型。這里列舉幾個(gè)常見(jiàn)的類型：

• stored： 字段原始內(nèi)容存儲(chǔ) 
• indexOptions:(NONE/DOCS/DOCS_AND_FREQS/DOCS_AND_FREQS_AND_POSITIONS/DOCS_AND_FREQS_AND_POSITIONS_AND_OFFSETS)，倒排索引的選項(xiàng)，存儲(chǔ)詞頻、位置信息等。
• docValuesType： 正排索引，建立一個(gè)docid到field的的一個(gè)列存儲(chǔ)。
• 一些其它的類型

IndexWriter：IndexWriter在doc進(jìn)行commit后，才會(huì)被持久化并且是可搜索的。IndexWriterConfig：IndexWriterConfig負(fù)責(zé)了一些整體的配置參數(shù)，并提供了方便使用者進(jìn)行功能定制的參數(shù)： 

• Similarity： 這個(gè)是搜索的核心參數(shù)，實(shí)現(xiàn)了這個(gè)接口就能夠進(jìn)行自定義算分。lucence默認(rèn)實(shí)現(xiàn)了前面文章提到的TF-IDF、BM25算法。 
• MergePolicy： 合并的策略。我們知道ElasticSearch會(huì)進(jìn)行合并，從而減少段的數(shù)量。 
• IndexerThreadPool： 線程池的管理。
• FlushPolicy： flush的策略。
• Analyzer： 定制分詞器。
• IndexDeletionPolicy： 提交管理。

PS：在ElasticSearch中，為了支持分布式的功能，新增了一些系統(tǒng)默認(rèn)字段：

• _uid，主鍵，在寫(xiě)入的時(shí)候，可以指定該Doc的ID值，如果不指定，則系統(tǒng)自動(dòng)生成一個(gè)唯一的UUID值。
• _version，版本字段，version來(lái)保證對(duì)文檔的變更正確的執(zhí)行，更新文檔時(shí)有用。 
• _source，原始信息，如果后面維護(hù)不需要reindex索引可以關(guān)閉該字段，從而節(jié)省空間 
• _routiong，路由字段。 
• 其它的字段

2.2. 并發(fā)模型

上面我們知道indexwriter負(fù)責(zé)了ElasticSearch索引增刪改查。那它具體是如何管理的呢？

2.2.1. 基本操作

關(guān)鍵點(diǎn)：  

• DocumentsWriter處理寫(xiě)請(qǐng)求，并分配具體的線程DocumentsWriterPerThread
• DocumentsWriterPerThread具有獨(dú)立內(nèi)存空間，對(duì)文檔進(jìn)行處理DocumentsWriter觸發(fā)一些flush的操作。
• DocumentsWriterPerThread中的內(nèi)存In-memory buffer會(huì)被flush成獨(dú)立的segement文件。 
• 對(duì)于這種設(shè)計(jì)，多線程的寫(xiě)入，針對(duì)純新增文檔的場(chǎng)景，所有數(shù)據(jù)都不會(huì)有沖突，非常適合隔離的數(shù)據(jù)寫(xiě)入方式

2.2.2 更新

Lucene的update和數(shù)據(jù)庫(kù)的update不太一樣，Lucene的更新是查詢后刪除再新增。  

• 分配一個(gè)操作線程 
• 在線程里執(zhí)行刪除 
• 在線程里執(zhí)行新增

2.2.3 刪除

上面已經(jīng)說(shuō)了，在update中會(huì)刪除，普通的也會(huì)刪除，lucence維護(hù)了一個(gè)全局的刪除表，每個(gè)線程也會(huì)維護(hù)一個(gè)刪除表，他們雙向同步數(shù)據(jù)

• update的刪除會(huì)先在內(nèi)部記錄刪除的數(shù)據(jù)，然后同步到全局表中。
• delete的刪除會(huì)作用在Global級(jí)別，后異步同步到線程中。
• Lucene Segment內(nèi)部，數(shù)據(jù)實(shí)際上其實(shí)并不會(huì)被真正刪除，Segment內(nèi)部會(huì)維持一個(gè)文件記錄，哪些是docid是刪除的，在merge時(shí)，相應(yīng)的doc文檔會(huì)被真正的刪除。

2.2.4 flush和commit

每一個(gè)WriterPerThread線程會(huì)根據(jù)flush策略將文檔形成segment文件，此時(shí)segment的文件還是不可見(jiàn)的，需要indexWriter進(jìn)行commit后才能被搜索。 這里需要注意：ElasticSearch的refresh對(duì)應(yīng)于lucene的flush，ElasticSearch的flush對(duì)應(yīng)于lucene的commit，ElasticSearch在refresh時(shí)通過(guò)其它方式使得segment變得可讀。

2.2.5 merge

merge是對(duì)segment文件合并的動(dòng)作，這樣可以提升查詢的效率并且可以真正的刪除的文檔。

小結(jié)

在這里我們稍微總結(jié)一下，一個(gè)ElasticSearch索引的一個(gè)分片對(duì)應(yīng)一個(gè)完整的lucene索引, 而一個(gè)lucene索引對(duì)應(yīng)多個(gè)segment。我們?cè)跇?gòu)建同一個(gè)lucene索引的時(shí)候, 可能有多個(gè)線程在并發(fā)構(gòu)建同一個(gè)lucene索引, 這個(gè)時(shí)候每個(gè)線程會(huì)對(duì)應(yīng)一個(gè)DocumentsWriterPerThread, 而每個(gè) DocumentsWriterPerThread會(huì)對(duì)應(yīng)一個(gè)index buffer. 在執(zhí)行了flush以后, 一個(gè) DocumentsWriterPerThread會(huì)生成一個(gè)segment。

三、 ElasticSearch的寫(xiě)

3.1. 宏觀看ElasticSearch請(qǐng)求

在前面的文章已經(jīng)討論了寫(xiě)入的流程ElasticSearch

圖片來(lái)自官網(wǎng) 當(dāng)寫(xiě)入文檔的時(shí)候，根據(jù)routing規(guī)則，會(huì)將文檔發(fā)送至特定的Shard中建立lucence。

• 介紹在Primary Shard上執(zhí)行成功后，再?gòu)腜rimary Shard上將請(qǐng)求同時(shí)發(fā)送給多個(gè)Replica Shardgit 
• 請(qǐng)求在多個(gè)Replica Shard上執(zhí)行成功并返回給Primary Shard后，寫(xiě)入請(qǐng)求執(zhí)行成功，返回結(jié)果給客戶端

注意上面的寫(xiě)入延時(shí)=主分片延時(shí)+max(Replicas Write),即寫(xiě)入性能如果有副本分片在，就至少是寫(xiě)入兩個(gè)分片的延時(shí)延時(shí)之和。

3.2. 詳細(xì)流程

3.2.1 協(xié)調(diào)節(jié)點(diǎn)內(nèi)部流程

如上圖所示：

• 協(xié)調(diào)節(jié)點(diǎn)會(huì)對(duì)請(qǐng)求檢查放在第一位，如果如果有問(wèn)題就直接拒絕。主要有長(zhǎng)度校驗(yàn)、必傳參數(shù)、類型、版本、id等等。
• pipeline，用戶可以自定義設(shè)置處理器，比如可以對(duì)字段切割或者新增字段，還支持一些腳本語(yǔ)言，可以查看官方文檔編寫(xiě)。
• 如果允許自動(dòng)創(chuàng)建索引（默認(rèn)是允許的），會(huì)先創(chuàng)建索引，創(chuàng)建索引會(huì)發(fā)送到主節(jié)點(diǎn)上，必須等待master成功響應(yīng)后，才會(huì)進(jìn)入下一流程。
• 請(qǐng)求預(yù)處理，比如是否會(huì)自動(dòng)生成id、路由，獲取到整個(gè)集群的信息了，并檢查集群狀態(tài)，比如集群master不存在，都會(huì)被拒絕。
• 構(gòu)建sharding請(qǐng)求，比如這一批有5個(gè)文檔， 如果都是屬于同一個(gè)分片的，那么就會(huì)合并到一個(gè)請(qǐng)求里，會(huì)根據(jù)路由算法將文檔分類放到一個(gè)map里 Map> requestsByShard = new HashMap<>();路由算法默認(rèn)是文檔id%分片數(shù)。
• 轉(zhuǎn)發(fā)請(qǐng)求，有了分片會(huì)根據(jù)前面的集群狀態(tài)來(lái)確定具體的ElasticSearch節(jié)點(diǎn)ip，然后并行去請(qǐng)求它們。

3.2.2 主分片節(jié)點(diǎn)流程*

 寫(xiě)入（index）

該部分是elasticsarch的核心寫(xiě)入流程，在前面的文章也介紹了，請(qǐng)求到該節(jié)點(diǎn)會(huì)最終調(diào)用lucence的方法，建立lucence索引。其中主要的關(guān)鍵點(diǎn)：

• ElasticSearch節(jié)點(diǎn)接收index請(qǐng)求，存入index buffer，同步存入磁盤translog后返回索引結(jié)果
• Refresh定時(shí)將lucence數(shù)據(jù)生成segment，存入到操作系統(tǒng)緩存，此時(shí)沒(méi)有fsync，清空l(shuí)ucence，此時(shí)就可以被ElasticSearch查詢了，如果index buffer占滿時(shí)，也會(huì)觸發(fā)refresh，默認(rèn)為jvm的10%。
• Flush定時(shí)將緩存中的segments寫(xiě)入到磁盤，刪除translog。如果translog滿時(shí)（512m），也會(huì)觸發(fā)flush。
• 如果數(shù)據(jù)很多，segment的也很多，同時(shí)也可能由刪除的文檔，ElasticSearch會(huì)定期將它們合并。

update

• 讀取同id的完整Doc, 記錄版本為version1。
• 將version1的doc和update請(qǐng)求的Doc合并成一個(gè)Doc，更新內(nèi)存中的VersionMap。獲取到完整Doc后。進(jìn)入后續(xù)的操作。
• 后面的操作會(huì)加鎖。
• 第二次從versionMap中讀取該doc的的最大版本號(hào)version2，這里基本都會(huì)從versionMap中獲取到。
• 檢查版本是否沖突，判斷版本是否一致（沖突），如果發(fā)生沖突，則回到第一步，重新執(zhí)行查詢doc合并操作。如果不沖突，則執(zhí)行最新的添加doc請(qǐng)求。
• 介紹在add Doc時(shí)，首先將Version + 1得到V3，再將Doc加入到Lucene中去，Lucene中會(huì)先刪同id下的已存在doc id，然后再增加新Doc。寫(xiě)入Lucene成功后，將當(dāng)前V3更新到versionMap中。
• 釋放鎖，更新流程就結(jié)束了。

介紹其實(shí)就是樂(lè)觀鎖的機(jī)制，每次更新一次版本號(hào)加 1 ，不像關(guān)系式數(shù)據(jù)庫(kù)有事物，你在更新數(shù)據(jù)，可能別人也在更新的話，就把你的給覆蓋了。你要更新的時(shí)候，先查詢出來(lái)，記住版本號(hào)，在更新的時(shí)候最新的版本號(hào)和你查詢的時(shí)候不一樣，說(shuō)明別人先更新了。你應(yīng)該讀取最新的數(shù)據(jù)之后再更新。寫(xiě)成功后，會(huì)轉(zhuǎn)發(fā)寫(xiě)副本分片，等待響應(yīng)，并最后返回?cái)?shù)據(jù)給協(xié)調(diào)節(jié)點(diǎn)。具體的流程：

• 校驗(yàn)，校驗(yàn)寫(xiě)的分片是否存在、索引的狀態(tài)是否正常等等。
• 是否需要延遲執(zhí)行，如果是則會(huì)放入到隊(duì)列里等待。
• 校驗(yàn)活躍的分片數(shù)是否存在，不足則拒絕寫(xiě)入。

public boolean enoughShardsActive(final int activeShardCount) {
  if (this.value < 0) {
    throw new IllegalStateException("not enough information to resolve to shard count");
  }
  if (activeShardCount < 0) {
    throw new IllegalArgumentException("activeShardCount cannot be negative");
  }
  return this.value <= activeShardCount;
}

為什么會(huì)要校驗(yàn)這個(gè)活躍的分片數(shù)呢？

• ElasticSearch的索引層有個(gè)一waitforactiveshards參數(shù)代表寫(xiě)入的時(shí)候必須的分片數(shù)，默認(rèn)是1。如果一個(gè)索引是每個(gè)分片3個(gè)副本的話，那么一共有4個(gè)分片，請(qǐng)求時(shí)至少需要校驗(yàn)存活的分片數(shù)至少為1，相當(dāng)于提前校驗(yàn)了。如果對(duì)數(shù)據(jù)的可靠性要求很高，就可以調(diào)高這個(gè)值，必須要達(dá)到這個(gè)數(shù)量才會(huì)寫(xiě)入。
• 調(diào)用lucence寫(xiě)入doc.
• 寫(xiě)入translog日志。
• 寫(xiě)入副本分片，循環(huán)處理副本請(qǐng)求，會(huì)傳遞一些信息。在這里需要注意，它們是異步發(fā)送到副本分片上的，并且需要全部等待響應(yīng)結(jié)果，直至超時(shí)。
• 接著上一步，如果有副本分片失敗的情況，會(huì)把這個(gè)失敗的分片發(fā)送給master，master會(huì)更新集群狀態(tài)，這個(gè)副本分片會(huì)從可分配列表中移除。 

發(fā)送請(qǐng)求至副本

@Override
public void tryAction(ActionListener<ReplicaResponse> listener) {
  replicasProxy.performOn(shard, replicaRequest, primaryTerm, globalCheckpoint, maxSeqNoOfUpdatesOrDeletes, listener);
}

等待結(jié)果

privatevoid decPendingAndFinishIfNeeded() {
  assert pendingActions.get() > 0 : "pending action count goes below 0 for request [" + request + "]";
  if (pendingActions.decrementAndGet() == 0) {
    finish();
  }
}

在以前的版本中，其實(shí)是異步請(qǐng)求副本分片的，后來(lái)覺(jué)得丟失數(shù)據(jù)的風(fēng)險(xiǎn)很大，就改成同步發(fā)送了，即Primary等Replica返回后再返回給客戶端。如果副本有寫(xiě)入失敗的，ElasticSearch會(huì)進(jìn)行一些重試，但最終并不強(qiáng)求一定要在多少個(gè)節(jié)點(diǎn)寫(xiě)入成功。在返回的結(jié)果中，會(huì)包含數(shù)據(jù)在多少個(gè)shard中寫(xiě)入成功了，多少個(gè)失敗了，如果有副本上傳失敗，會(huì)將失敗的副本上報(bào)至Master。

PS：ElasticSearch的數(shù)據(jù)副本模型和kafka副本很相似，都是采用的是ISR機(jī)制。即：ES里面有一個(gè)：in-sync copies概念，主分片會(huì)在索引的時(shí)候會(huì)同步數(shù)據(jù)至in-sync copies里面所有的節(jié)點(diǎn)，然后再返回ACK給client。而in-sync copies里面的節(jié)點(diǎn)是動(dòng)態(tài)變化的，如果出現(xiàn)極端情況，在in-sync copies列表中只有主分片一個(gè)的話，這里很容易出現(xiàn)SPOF問(wèn)題，這個(gè)是在ElasticSearch中是如何解決的呢？

就是依靠上面我們分析的wait_for_active_shards參數(shù)來(lái)防止SPOF，如果配置index的wait_for_active_shards=3就會(huì)提前校驗(yàn)必須要有三個(gè)活躍的分片才會(huì)進(jìn)行同步，否則拒絕請(qǐng)求。對(duì)于可靠性要求高的索引可以提升這個(gè)值。

PS：為什么是先寫(xiě)lucence再寫(xiě)入translog呢，這是因?yàn)閷?xiě)入lucence寫(xiě)入時(shí)會(huì)有數(shù)據(jù)檢查，有可能會(huì)寫(xiě)入失敗，這個(gè)是發(fā)生在內(nèi)存之中的，如果先寫(xiě)入磁盤的translog的話，還需要回退日志，比較麻煩

3.2.3 副本分片節(jié)點(diǎn)流程8

這個(gè)過(guò)程和主分片節(jié)點(diǎn)的流程基本一樣，有些校驗(yàn)可能略微不同，最終都會(huì)寫(xiě)入lucence索引。

四、總結(jié)

本文介紹了ElasticSearch的寫(xiě)入流程和一些比較詳細(xì)的機(jī)制，最后我們總結(jié)下開(kāi)頭我們提出的問(wèn)題，一個(gè)分布式系統(tǒng)需要滿足很多特性，大部分特性都能夠在ElasticSearch中得到滿足。

• 可靠性：lucence只是個(gè)工具，ElasticSearch中通過(guò)自己設(shè)計(jì)的副本來(lái)保證了節(jié)點(diǎn)的容錯(cuò)，通過(guò)translog日志保證宕機(jī)后能夠恢復(fù)。通過(guò)這兩套機(jī)制提供了可靠性保障。
• 一致性：ElasticSearch實(shí)現(xiàn)的是最終一致性，副本和主分片在同一時(shí)刻讀取的數(shù)據(jù)可能不一致。比如副本的refresh頻率和主分片的頻率可能不一樣。
• 高性能：ElasticSearch通過(guò)多種手段來(lái)提升性能，具體包括：

• lucence自身獨(dú)立線程維護(hù)各自的Segment，多線程需要競(jìng)爭(zhēng)的資源更少，性能更好。 
• update等操作使用versionMap緩存，減少io.
• refresh至操作系統(tǒng)緩存。

• 原子性、隔離性：使用版本的樂(lè)觀鎖機(jī)制保證的。
• 實(shí)時(shí)性：ElasticSearch設(shè)計(jì)的是近實(shí)時(shí)的，如果同步進(jìn)行refresh、flush將大幅降低性能，所以是”攢一部分?jǐn)?shù)據(jù)“再刷入磁盤，不過(guò)實(shí)時(shí)寫(xiě)入的tranlog日志還是可以實(shí)時(shí)通過(guò)id查到的。

以上就是ElasticSearch寫(xiě)入流程實(shí)例解析的詳細(xì)內(nèi)容，更多關(guān)于ElasticSearch寫(xiě)入流程的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！