1. 背景

????????Elasticsearch 是一個(gè)實(shí)時(shí)的分布式搜索分析引擎，簡稱 ES。一個(gè)集群由多個(gè)節(jié)點(diǎn)組成，節(jié)點(diǎn)的角色可以根據(jù)用戶的使用場景自由配置，集群可以以節(jié)點(diǎn)為單位自由擴(kuò)縮容，數(shù)據(jù)以索引、分片的形式散列在各個(gè)節(jié)點(diǎn)上。本文介紹 ES 分布式架構(gòu)基礎(chǔ)原理，剖析分布式元數(shù)據(jù)管理模型，并介紹騰訊云 ES 在分布式擴(kuò)展性層面相關(guān)的優(yōu)化，解析代碼基于 8.5 版本。

2. 分布式架構(gòu)

????????我們首先來看看 ES 的分布式架構(gòu)。一個(gè) ES 集群包含多個(gè)節(jié)點(diǎn)，節(jié)點(diǎn)的類型有多種，最新版本主要的角色類型包括：

• master 節(jié)點(diǎn)：可設(shè)置專屬 master 節(jié)點(diǎn)，也可以和其它節(jié)點(diǎn)角色共享。
• 數(shù)據(jù)節(jié)點(diǎn)： data、data_hot、data_warm、data_cold、data_frozen 等。
• 功能節(jié)點(diǎn)： ingest、ml、remote_cluster_client、transform 等。

????????節(jié)點(diǎn)角色分為三類：主節(jié)點(diǎn)（master）用于管理集群；數(shù)據(jù)節(jié)點(diǎn)（data）用于存儲(chǔ)管理數(shù)據(jù)，數(shù)據(jù)節(jié)點(diǎn)按照數(shù)據(jù)特點(diǎn)分為多種類型，冷溫?zé)岬鹊?；功能?jié)點(diǎn)，包括 ETL、機(jī)器學(xué)習(xí)、遠(yuǎn)程管理等等。

下面我們拿一種典型的大規(guī)模集群場景架構(gòu)來舉例說明。

ES 分布式架構(gòu)

????????上圖所示的分布式架構(gòu)中，上面部分是專屬 master 節(jié)點(diǎn)，負(fù)責(zé)管理集群的元數(shù)據(jù)，他們之間通過基于類 Raft 的分布式一致性協(xié)議進(jìn)行選主、元數(shù)據(jù)同步。下面部分是專屬 data 節(jié)點(diǎn)，data 節(jié)點(diǎn)上包含索引分片。索引對應(yīng)傳統(tǒng)關(guān)系數(shù)據(jù)庫的表，是邏輯概念，一個(gè)索引包含多個(gè)分片，分片是節(jié)點(diǎn)上的數(shù)據(jù)存儲(chǔ)單元，它按照主鍵 hash 或用戶自定義數(shù)據(jù)路由（routing）均勻分布到各個(gè)節(jié)點(diǎn)。分片一般包含一主（primary）、多從（replica）分片，一個(gè)索引主從分片之間的數(shù)據(jù)復(fù)制模型基于微軟提出的 PacificA 協(xié)議。

3. 讀寫模型

3.1 寫入模型

????????ES 的任意節(jié)點(diǎn)可作為寫入請求的協(xié)調(diào)節(jié)點(diǎn)，接收用戶請求，協(xié)調(diào)節(jié)點(diǎn)先將寫入請求 hash 至分片粒度并先轉(zhuǎn)發(fā)對應(yīng)主分片寫入，主分片寫入成功再轉(zhuǎn)發(fā)至從分片，主從分片均寫入完畢經(jīng)協(xié)調(diào)節(jié)點(diǎn)返回客戶端成功。

????????騰訊云 ES 內(nèi)核團(tuán)隊(duì)實(shí)現(xiàn)了寫入分組定向路由、主從分片物理復(fù)制能力，可以減少從分片的寫入棧計(jì)算開銷，寫入性能提升50%+。

分布式寫入模型

????????本文不詳細(xì)解析寫入過程，有興趣同學(xué)可參考寫入流程解析：https://cloud.tencent.com/developer/article/1370501

3.2 查詢模型

????????和寫入一樣，ES 的任意節(jié)點(diǎn)可以作為查詢請求的協(xié)調(diào)節(jié)點(diǎn)，請求轉(zhuǎn)發(fā)至對應(yīng)一個(gè)或多個(gè)（取決于路由規(guī)則，不指定路由默認(rèn)索引所有分片均查詢）數(shù)據(jù)分片的主或者從分片進(jìn)行查詢，查詢根據(jù)復(fù)雜度分不同類型，QUERY_THEN_FETCH（兩階段），QUERY_AND_FETCH（一階段），各個(gè)分片查詢結(jié)果最后在協(xié)調(diào)節(jié)點(diǎn)匯聚，返回最終結(jié)果給客戶端。

分布式查詢模型

????????本文不詳細(xì)解析查詢過程，有興趣同學(xué)可參考查詢流程解析：https://cloud.tencent.com/developer/article/1154813

4. 分布式架構(gòu)元數(shù)據(jù)模型分類

????????在大規(guī)模分布式存儲(chǔ)架構(gòu)中，元數(shù)據(jù)模型主要分為中心化架構(gòu)和對等架構(gòu)兩類。中心化架構(gòu)的典型代表包括 HDFS、ES、BigTable 等，對等架構(gòu)典型的系統(tǒng)包括 Cassandra、Dynamo 等。

4.1 中心化架構(gòu)

????????中心化架構(gòu)的特點(diǎn)是專有的主節(jié)點(diǎn)管理集群元數(shù)據(jù)，HDFS 中 Namenode 統(tǒng)一管理元數(shù)據(jù)，data node 不單獨(dú)維護(hù)，客戶端找 Namenode 獲取路由信息。中心化架構(gòu)的優(yōu)勢是在大規(guī)模集群場景下，元數(shù)據(jù)同步范圍收斂，效率更高。由于主節(jié)點(diǎn)提供路由查詢信息，因此其主要缺點(diǎn)是主節(jié)點(diǎn)易成為瓶頸，一般通過 Federation 機(jī)制優(yōu)化，或者將元數(shù)據(jù)存儲(chǔ)到數(shù)據(jù)表中分布在部分節(jié)點(diǎn)上管理。ES 也采用了中心化架構(gòu)，稍后我們展開介紹。

中心化架構(gòu)

4.2 對等架構(gòu)

????????對等架構(gòu)的特點(diǎn)是去中心化，沒有獨(dú)立的主節(jié)點(diǎn)。節(jié)點(diǎn)之間通過 Gossip 協(xié)議傳播同步元數(shù)據(jù)，所有節(jié)點(diǎn)保存全量的元數(shù)據(jù)。這種方式架構(gòu)簡單清晰，沒有中心化瓶頸。缺點(diǎn)是通過 Gossip 協(xié)議收斂元數(shù)據(jù)效率偏低，受節(jié)點(diǎn)數(shù)量限制，擴(kuò)展性弱。

對等架構(gòu)

5. 元數(shù)據(jù)模型

????????前面我們介紹了 ES 的分布式架構(gòu)基礎(chǔ)原理及讀寫基本流程，也了解了業(yè)內(nèi)常用的分布式架構(gòu)元數(shù)據(jù)管理模型。下面我們來看看 ES 是如何管理集群的，其核心元數(shù)據(jù)模型是如何運(yùn)作的。

????????ES 的元數(shù)據(jù)由 master 節(jié)點(diǎn)維護(hù)管理，同時(shí)其它節(jié)點(diǎn)也維護(hù)著全量的元數(shù)據(jù)，目的是為了確保每個(gè)節(jié)點(diǎn)都能承擔(dān)數(shù)據(jù)路由的能力。元數(shù)據(jù)包括節(jié)點(diǎn)信息、索引信息、分片路由信息、配置信息等等，下面我們先揭開元數(shù)據(jù)在內(nèi)存中的神秘面紗，然后再看看元數(shù)據(jù)是如何持久化的。

5.1 內(nèi)存結(jié)構(gòu)

????????ES 元數(shù)據(jù)的內(nèi)存數(shù)據(jù)結(jié)構(gòu)對象是 ClusterState，代碼層面對應(yīng)的是 ClusterState.java。其主要的成員包含：

????????上面的內(nèi)存結(jié)構(gòu)中，對象占比最大的是 nodes、metaData、routingTable 和 routingNodes，這四塊是元數(shù)據(jù)的核心組成部分：

元數(shù)據(jù)核心結(jié)構(gòu)

????????接下來對這四個(gè)部分的內(nèi)部結(jié)構(gòu)進(jìn)行展開分析。

5.1.1 DiscoveryNodes

DiscoveryNodes 對象描述的是集群的節(jié)點(diǎn)信息。里面包含如下主要的成員信息：

// 當(dāng)前 master
private final String masterNodeId;
// 本地節(jié)點(diǎn)
private final String localNodeId;
// 完整的節(jié)點(diǎn)列表
private final ImmutableOpenMap<String, DiscoveryNode> nodes;

// 其它分類型節(jié)點(diǎn)列表
private final ImmutableOpenMap<String, DiscoveryNode> dataNodes;
private final ImmutableOpenMap<String, DiscoveryNode> masterNodes;
private final ImmutableOpenMap<String, DiscoveryNode> ingestNodes;

復(fù)制

????????從變量名稱很容易理解，其包含了本地節(jié)點(diǎn)、當(dāng)前 master 節(jié)點(diǎn)的 id 信息，以及集群全量節(jié)點(diǎn)列表（nodes），并將這個(gè)列表分類為不同類型的節(jié)點(diǎn)列表。

5.1.2 MetaData

MetaData 包含集群全局的配置信息。主要的成員包括：

// 集群的動(dòng)態(tài)設(shè)定，transient 全量重啟后就沒有了，persistent 會(huì)持久化
private final Settings transientSettings;
private final Settings persistentSettings;

// 索引、模板信息
private final ImmutableOpenMap<String, IndexMetaData> indices;
private final ImmutableOpenMap<String, IndexTemplateMetaData> templates;

// 集群選舉管理信息
private final CoordinationMetaData coordinationMetaData;

// raft 選舉使用的 term
private final long term;

// 最近提交的選舉節(jié)點(diǎn)列表，內(nèi)部是一個(gè) Set<String> nodeIds
private final VotingConfiguration lastCommittedConfiguration;

// 最近接收的選舉節(jié)點(diǎn)列表
private final VotingConfiguration lastAcceptedConfiguration;

// 用戶通過 _cluster/voting_config_exclusions 接口設(shè)定的選舉排除節(jié)點(diǎn)列表
private final Set<VotingConfigExclusion> votingConfigExclusions;

復(fù)制

5.1.3 RoutingTable

????????RoutingTable 包含主要的數(shù)據(jù)路由信息，在查詢、寫入請求的時(shí)候提供分片到節(jié)點(diǎn)的路由，動(dòng)態(tài)構(gòu)建不會(huì)持久化。我們先來通過一張圖了解這個(gè)數(shù)據(jù)路由結(jié)構(gòu)的包含關(guān)系：

分片到節(jié)點(diǎn)的映射（路由表）

????????一個(gè) RoutingTable 包含集群的所有索引，一個(gè)索引包含多個(gè)分片，一個(gè)分片包含一個(gè)主、多個(gè)從本分片，最底層的 ShardRouting 描述具體的某個(gè)副本分片所在的節(jié)點(diǎn)以及正在搬遷的目標(biāo)節(jié)點(diǎn)信息。用戶請求時(shí)只指定索引信息，請求到達(dá)協(xié)調(diào)節(jié)點(diǎn)，由協(xié)調(diào)節(jié)點(diǎn)根據(jù)該路由表來獲取底層分片所在節(jié)點(diǎn)并轉(zhuǎn)發(fā)請求。

????????接下來看看對應(yīng)的數(shù)據(jù)結(jié)構(gòu)，RoutingTable 頂層主要的成員只有一個(gè)：

// key 為索引名，IndexRoutingTable 為該索引包含的分片信息
private final ImmutableOpenMap<String, IndexRoutingTable> indicesRouting;

復(fù)制

????????IndexRoutingTable 對象：

// 索引信息，包含索引名稱、uuid 等。
private final Index index;

// 索引的分片列表，key 為分片的編號(hào)，例如 2 號(hào)分片，3 號(hào)分片
private final ImmutableOpenIntMap<IndexShardRoutingTable> shards;

復(fù)制

????????一套分片信息包含一個(gè)主、多個(gè)從分片，其對象結(jié)構(gòu) IndexShardRoutingTable：

// 當(dāng)前分片信息，主要包含分片的編號(hào)、分片對應(yīng)的索引信息
final ShardId shardId;

// 當(dāng)前分片的主分片信息
final ShardRouting primary;

// 當(dāng)前分片的多個(gè)從分片列表
final List<ShardRouting> replicas;

// 當(dāng)前分片全量的分片列表，包含主、副本
final List<ShardRouting> shards;

// 當(dāng)前分片已經(jīng) started 的活躍分片列表
final List<ShardRouting> activeShards;

// 當(dāng)前分片已經(jīng)分配了節(jié)點(diǎn)的分片列表
final List<ShardRouting> assignedShards;

復(fù)制

分片最底層的數(shù)據(jù)結(jié)構(gòu)是 ShardRouting ，它描述這個(gè)分片的狀態(tài)、節(jié)點(diǎn)歸屬等信息：

private final ShardId shardId;

// 分片所在的當(dāng)前節(jié)點(diǎn) id
private final String currentNodeId;

// 如果分片正在搬遷，這里為目標(biāo)節(jié)點(diǎn) id
private final String relocatingNodeId;

// 是否是主分片
private final boolean primary;

// 分片的狀態(tài)，UNASSIGNED/INITIALIZING/STARTED/RELOCATING
private final ShardRoutingState state;

// 每一個(gè)分片分配后都有一個(gè)唯一標(biāo)識(shí)
private final AllocationId allocationId;

復(fù)制

5.1.4 RoutingNodes

????????該對象為節(jié)點(diǎn)到分片的映射關(guān)系。主要用于統(tǒng)計(jì)每個(gè)節(jié)點(diǎn)的分片分配情況，在分片分配、均衡的時(shí)候使用，需要時(shí)根據(jù) RoutingTable 動(dòng)態(tài)構(gòu)建，不會(huì)持久化。

節(jié)點(diǎn)到分片的映射

RoutingNodes 的主要成員包括：

// 節(jié)點(diǎn)到分片的映射，key 為 nodeId，value 為一個(gè)包含分片列表的節(jié)點(diǎn)信息
private final Map<String, RoutingNode> nodesToShards = new HashMap<>();

// 未分配的分片列表，每次 reroute 從這里獲取分片分配
private final UnassignedShards unassignedShards = new UnassignedShards(this);

// 已經(jīng)分配的分片列表
private final Map<ShardId, List<ShardRouting>> assignedShards = new HashMap<>();

// 當(dāng)前分片遠(yuǎn)程恢復(fù)的統(tǒng)計(jì)數(shù)據(jù) incoming/outcoming
private final Map<String, Recoveries> recoveriesPerNode = new HashMap<>();

復(fù)制

上面 RoutingNode 的主要結(jié)構(gòu)包括：

private final String nodeId;

// 該節(jié)點(diǎn)上的分片列表
private final LinkedHashMap<ShardId, ShardRouting> shards; 

// 該節(jié)點(diǎn)上的初始化中的分片列表
private final LinkedHashSet<ShardRouting> initializingShards;

// 該節(jié)點(diǎn)上的正在搬遷（搬走）的分片列表
private final LinkedHashSet<ShardRouting> relocatingShards;

復(fù)制

????????前面就是所有元數(shù)據(jù)在內(nèi)存中的結(jié)構(gòu)，最后我們借助這張 UML 圖來看元數(shù)據(jù)的整體結(jié)構(gòu)和關(guān)系：

元數(shù)據(jù)類結(jié)構(gòu)

5.2 元數(shù)據(jù)持久化

????????前面介紹的元數(shù)據(jù)內(nèi)存結(jié)構(gòu)中，只有部分信息會(huì)被持久化存儲(chǔ)到磁盤上，其它結(jié)構(gòu)都是在節(jié)點(diǎn)啟動(dòng)后動(dòng)態(tài)構(gòu)建或者直接在內(nèi)存中動(dòng)態(tài)維護(hù)的。持久化的內(nèi)容包含兩部分，索引元數(shù)據(jù)和集群元數(shù)據(jù)也叫全局元數(shù)據(jù)，下面我們分別來介紹這兩部分持久化的內(nèi)容。

5.2.1 索引元數(shù)據(jù)（index metadata）

索引元數(shù)據(jù)維護(hù)的是各個(gè)索引獨(dú)有的配置信息，持久化的內(nèi)容主要包括：

• in_sync_allocations：每個(gè)分片分配之后都有一個(gè)唯一的 allocationId，該列表是主分片用來維護(hù)被認(rèn)為跟自己保持?jǐn)?shù)據(jù)一致的副本列表。從這個(gè)列表中踢出的分片不會(huì)接收新的寫入，需要走分片恢復(fù)流程將數(shù)據(jù)追齊之后才能再被放入該隊(duì)列。
• mappings：索引的 mapping 信息，定義各個(gè)字段的類型、屬性。
• settings：索引自己的配置信息。
• state：索引狀態(tài)，OPEN/CLOSED。
• aliases：索引別名信息。
• routing_num_shards：索引分片數(shù)量。
• primary_terms：每一輪分片切主都會(huì)產(chǎn)生新的 primary term，用于保持主從分片之間的一致性。

5.2.2 全局元數(shù)據(jù)（global metadata）

????????全局元數(shù)據(jù)持久化的內(nèi)容主要是前面描述的 MetaData 對象中去除 indices 索引部分。包括動(dòng)態(tài)配置、模板信息、選舉信息等。主要包含三部分：

• manifest-數(shù)字.st：該文件是一個(gè)磁盤元數(shù)據(jù)管理入口，主要管理元數(shù)據(jù)的版本信息。
• global-數(shù)字.st：MetaData 中的全局元數(shù)據(jù)的主要信息就是持久化到這個(gè)文件，包括動(dòng)態(tài)配置、模板信息等。
• node-數(shù)字.st：當(dāng)前節(jié)點(diǎn)的元數(shù)據(jù)信息，包括節(jié)點(diǎn)的 nodeId 和版本信息。

5.2.3 元數(shù)據(jù)文件分布

????????在 7.6.0 版本之前，元數(shù)據(jù)直接存放在每個(gè)節(jié)點(diǎn)磁盤上，且在專有 master 節(jié)點(diǎn)上每個(gè)索引一個(gè)元數(shù)據(jù)目錄。在 7.6.0 版本之后，ES 將元數(shù)據(jù)放到節(jié)點(diǎn)本地獨(dú)立的 lucene 索引中保存。7.6.0 之前持久化的數(shù)據(jù)目錄包含的內(nèi)容：

├── indices
│   ├── 00IUbRWZSzGsN71k0TlPmA
│   │   └── _state
│   │   └── state-8.st
│   ├── 01RTGqCbRe-9PYO55j8zAQ
│   │   └── _state
│   │   └── state-10.st
│   ├── 02DSrGNzRzizuI42Yf6aJg
│   │   └── _state
│   │   └── state-17.st

├── node.lock
└── _state
├── global-4928.st
└── node-0.st

復(fù)制

????????上圖中上半部分是每個(gè)索引的元數(shù)據(jù)，它們位于 indices 目錄下，一個(gè)索引對應(yīng)一個(gè) uuid 目錄，下面有一個(gè) _state 文件夾，索引自己的元數(shù)據(jù)就放在這個(gè) _state 下面。下半部分就是我們上面描述的全局元數(shù)據(jù)持久化的文件。

????????7.6.0 之后 data/nodes/0 里面保存的是 segment 文件，元數(shù)據(jù)以本地 Lucene index 方式持久化，收斂文件數(shù)量。其中 node-0.st 是舊版升級后的兼容文件：

├── node.lock
└── _state
├── _3q.cfe
├── _3q.cfs
├── _3q.si
├── _3r.cfe
├── _3r.cfs
├── _3r.si
├── node-0.st
├── segments_6i
└── write.lock

復(fù)制

元數(shù)據(jù)索引同時(shí)保存全局、索引元數(shù)據(jù)，用 type 字段區(qū)分，索引包含三個(gè)字段：

• type：cluster 級別元數(shù)據(jù) “global”；索引級別元數(shù)據(jù) “index”。
• index_uuid：索引元數(shù)據(jù)對應(yīng)索引的 UUID。
• data：metadata 元數(shù)據(jù)內(nèi)容。

同時(shí)，在每次提交保存的時(shí)候，會(huì)存一份 commit user data，包括選舉的 term，最新的版本、node id、node version 等信息：

* +------------------------------+-----------------------------+----------------------------------------------+
* | "type" (string field)        | "index_uuid" (string field) | "data" (stored binary field in SMILE format) |
* +------------------------------+-----------------------------+----------------------------------------------+
* | GLOBAL_TYPE_NAME == "global" | (omitted)                   | Global metadata                              |
* | INDEX_TYPE_NAME  == "index"  | Index UUID                  | Index metadata                               |
* +------------------------------+-----------------------------+----------------------------------------------+
*
* Additionally each commit has the following user data:
*
* +---------------------------+-------------------------+-------------------------------------------------------------------------------+
* |        Key symbol         |       Key literal       |                                     Value                                     |
* +---------------------------+-------------------------+-------------------------------------------------------------------------------+
* | CURRENT_TERM_KEY          | "current_term"          | Node's "current" term (≥ last-accepted term and the terms of all sent joins)  |
* | LAST_ACCEPTED_VERSION_KEY | "last_accepted_version" | The cluster state version corresponding with the persisted metadata           |
* | NODE_ID_KEY               | "node_id"               | The (persistent) ID of the node that wrote this metadata                      |
* | NODE_VERSION_KEY          | "node_version"          | The (ID of the) version of the node that wrote this metadata                  |
* +---------------------------+-------------------------+-------------------------------------------------------------------------------+

復(fù)制

6. 元數(shù)據(jù)管理

????????ES 的索引創(chuàng)建、刪除、mapping 更新、template 更新、節(jié)點(diǎn)加入脫離等 DDL 操作都會(huì)涉及到元數(shù)據(jù)變更。前面我們從內(nèi)存、持久化層面介紹了 ES 元數(shù)據(jù)的組成部分，接下來我們看看 ES 是如何對元數(shù)據(jù)進(jìn)行管理的。

6.1 元數(shù)據(jù)初始化

????????ES 節(jié)點(diǎn)的啟動(dòng)入口在 Node.java 的 start 函數(shù)中，里面會(huì)初始化各種 service。元數(shù)據(jù)管理相關(guān)的 service 也是在這個(gè)環(huán)節(jié)進(jìn)行初始化。GatewayMetaState 對象負(fù)責(zé)元數(shù)據(jù)的存取，在節(jié)點(diǎn)啟動(dòng)過程中會(huì)根據(jù)節(jié)點(diǎn)的類型，確定元數(shù)據(jù)的存取方式。分為以下幾種場景：

• master 節(jié)點(diǎn)。基于 LucenePersistedState 對象同步落盤，每當(dāng)節(jié)點(diǎn)收到新的元數(shù)據(jù)的時(shí)候會(huì)馬上保存到磁盤。并更新緩存，讀取的時(shí)候優(yōu)先讀取緩存對象，節(jié)點(diǎn)啟動(dòng)的時(shí)候緩存會(huì)從磁盤加載。
• data 節(jié)點(diǎn)。包括前述 data 前綴的 role，基于 AsyncPersistedState 對象異步落盤，對應(yīng)異步線程名稱是 AsyncLucenePersistedState#updateTask。
• 其它功能節(jié)點(diǎn)。非 master 屬性且不保存數(shù)據(jù)的節(jié)點(diǎn)，例如 ingest、ml、transform 等。基于 InMemoryPersistedState 對象直接保存到內(nèi)存不做持久化。無論是在內(nèi)存中的還是持久化的元數(shù)據(jù)對象，它們都對外暴露一個(gè) PersistedState 接口。提供保存最后接收的元數(shù)據(jù)（ setLastAcceptedState），以及獲取最后接收的元數(shù)據(jù)（getLastAcceptedState）。

GatewayMetaState.java 代碼片段：

if (DiscoveryNode.isMasterNode(settings) || DiscoveryNode.canContainData(settings)) {
    ......
    if (DiscoveryNode.isMasterNode(settings)) {
        persistedState = new LucenePersistedState(persistedClusterStateService, currentTerm, clusterState);
    } else {
        persistedState = new AsyncPersistedState(
            settings,
            transportService.getThreadPool(),
            new LucenePersistedState(persistedClusterStateService, currentTerm, clusterState)
        );
    }        
    ......
    this.persistedState.set(persistedState);
} else {
    ......
    persistedState.set(new InMemoryPersistedState(currentTerm, clusterState));
}

復(fù)制

6.2 發(fā)布流程

????????接下來我們以最典型的 DDL 創(chuàng)建索引操作為例，介紹整個(gè)元數(shù)據(jù)發(fā)布流程。ES 的 DDL 操作都是通過元數(shù)據(jù)變更推導(dǎo)模式實(shí)現(xiàn)的，例如創(chuàng)建索引，首先 master 會(huì)產(chǎn)生一版帶新增索引的元數(shù)據(jù)，并將該新版元數(shù)據(jù)發(fā)布至各個(gè)節(jié)點(diǎn)，各個(gè)節(jié)點(diǎn)和自己上一個(gè)持久化的元數(shù)據(jù)版本進(jìn)行比對，產(chǎn)生差異化的索引執(zhí)行創(chuàng)建索引、分片的任務(wù)。

下圖是一張宏觀的索引創(chuàng)建元數(shù)據(jù)變更流程，方便大家從整體架構(gòu)上有個(gè)初步的了解。

索引創(chuàng)建元數(shù)據(jù)變更流程

????????接下來分別從 master、data 節(jié)點(diǎn)維度分別看著兩階段的處理流程。下面是兩個(gè)階段的各個(gè)狀態(tài)：

enum PublicationTargetState {
    NOT_STARTED,
    FAILED,
    SENT_PUBLISH_REQUEST, // 已發(fā)送 publish 請求
    WAITING_FOR_QUORUM,   // 等待大多數(shù)節(jié)點(diǎn)響應(yīng)
    SENT_APPLY_COMMIT,    // 已發(fā)送 commit 請求
    APPLIED_COMMIT,       // 元數(shù)據(jù)應(yīng)用完畢
}

復(fù)制

6.2.1 Master publish 階段

????????用戶發(fā)起 create index 請求，會(huì)先到達(dá) rest 層，由 RestCreateIndexAction 解析請求，并產(chǎn)生 transport 層的請求轉(zhuǎn)發(fā)給 master 處理。調(diào)用鏈：

RestCreateIndexAction -> TransportCreateIndexAction -> MetadataCreateIndexService

所有參數(shù)解析完畢之后，就進(jìn)入提交集群元數(shù)據(jù)變更任務(wù)流程：

private void onlyCreateIndex(final CreateIndexClusterStateUpdateRequest request, final ActionListener<AcknowledgedResponse> listener) {
    normalizeRequestSetting(request);
    //提交元數(shù)據(jù)變更任務(wù)，索引創(chuàng)建優(yōu)先級為 URGENT
    submitUnbatchedTask(
        "create-index [" + request.index() + "], cause [" + request.cause() + "]",
        new AckedClusterStateUpdateTask(Priority.URGENT, request, listener) {

            @Override
            public ClusterState execute(ClusterState currentState) throws Exception {
                // 分配分片產(chǎn)生新版本的元數(shù)據(jù)
                return applyCreateIndexRequest(currentState, request, false);
            }

            @Override
            public void onFailure(Exception e) {
                ......
            }
        }
    );
}

復(fù)制

????????代碼中 applyCreateIndexRequest 主要負(fù)責(zé)索引創(chuàng)建過程中，分配分片產(chǎn)生新版本的元數(shù)據(jù)，里面涉及復(fù)雜的分片分配、均衡策略流程，騰訊云 ES 內(nèi)核結(jié)合單個(gè)索引分片數(shù)、節(jié)點(diǎn)主分片數(shù)、節(jié)點(diǎn)總分片數(shù)、節(jié)點(diǎn)存儲(chǔ)空間多個(gè)維度深度定制優(yōu)化了均衡策略，徹底解決社區(qū)版各個(gè)維度負(fù)載不均的問題，這里不展開，以后再單獨(dú)介紹。

????????上述任務(wù)經(jīng)過 MasterService.submitStateUpdateTask 包裝之后提交給線程池執(zhí)行：

public void submitTask(BatchedTask task, @Nullable TimeValue timeout) throws EsRejectedExecutionException {
    tasksPerBatchingKey.compute(task.batchingKey, (k, existingTasks) -> {
        if (existingTasks == null) {
            existingTasks = Collections.synchronizedSet(new LinkedHashSet<>());
        } else {
            assert assertNoDuplicateTasks(task, existingTasks);
        }
        // 按 executor 匯總 task
        existingTasks.add(task);
        return existingTasks;
    });

    if (timeout != null) {
        // 指定超時(shí)時(shí)間，不指定默認(rèn)元數(shù)據(jù)變更 30s 超時(shí)
        threadExecutor.execute(task, timeout, () -> onTimeoutInternal(task, timeout));
    } else {
        threadExecutor.execute(task);
    }
}

復(fù)制

????????元數(shù)據(jù)任務(wù)支持 batch 方式執(zhí)行，相同類別的任務(wù)例如 template 變更任務(wù)可以批量變更，參考 MetadataIndexTemplateService.TEMPLATE_TASK_EXECUTOR。創(chuàng)建索引任務(wù)不能批量執(zhí)行。元數(shù)據(jù)任務(wù)執(zhí)行采用的是單線程多任務(wù)按優(yōu)先級串行執(zhí)行的模式，線程名稱為 masterService#updateTask，上述 threadExecutor 的初始化過程：

protected PrioritizedEsThreadPoolExecutor createThreadPoolExecutor() {
    return EsExecutors.newSinglePrioritizing(
        // masterService#updateTask，我們經(jīng)常在 jstack 堆棧中看到的 master 節(jié)點(diǎn)元數(shù)據(jù)變更線程
        nodeName + "/" + MASTER_UPDATE_THREAD_NAME,
        daemonThreadFactory(nodeName, MASTER_UPDATE_THREAD_NAME),
        threadPool.getThreadContext(),
        threadPool.scheduler(),
        ......
    );
}

復(fù)制

元數(shù)據(jù)變更的優(yōu)先級：

public enum Priority {
    IMMEDIATE((byte) 0), // 節(jié)點(diǎn)加入、脫離等
    URGENT((byte) 1), // 索引、template 創(chuàng)建等
    HIGH((byte) 2),
    NORMAL((byte) 3),
    LOW((byte) 4),
    LANGUID((byte) 5);
}

復(fù)制

元數(shù)據(jù)變更任務(wù)產(chǎn)生了新的元數(shù)據(jù)之后，就會(huì)進(jìn)入元數(shù)據(jù) publish 階段，調(diào)用棧：

MasterService.publishClusterStateUpdate() -> MasterService.publish() -> Coordinator.publish()

核心邏輯在 Coordinator 中處理，該類負(fù)責(zé)管理元數(shù)據(jù)的 publish、commit 流程。publish 函數(shù)的代碼片段：

@Override
public void publish(
    ClusterStatePublicationEvent clusterStatePublicationEvent,
    ActionListener<Void> publishListener,
    AckListener ackListener
) {
    try {
        synchronized (mutex) {
           ......
            try {
                ......
                // 獲取目標(biāo)發(fā)布節(jié)點(diǎn)
                final DiscoveryNodes publishNodes = publishRequest.getAcceptedState().nodes();
                // leader checker 在數(shù)據(jù)節(jié)點(diǎn)上用于追蹤 master，和 master 保持心跳
                leaderChecker.setCurrentNodes(publishNodes);
                // 設(shè)置當(dāng)前 leader 需要追蹤的 follower 節(jié)點(diǎn)，主要是用于心跳保持
                followersChecker.setCurrentNodes(publishNodes);
                // lag 探測器，如果某個(gè)節(jié)點(diǎn)超過指定時(shí)間默認(rèn) 90s 沒有 publish 成功則踢出
                lagDetector.setTrackedNodes(publishNodes);
                // 啟動(dòng) publish 任務(wù)
                publication.start(followersChecker.getFaultyNodes());
            } finally {
                publicationContext.decRef();
            }
        }
    } catch (Exception e) {
       ......
    }
}

復(fù)制

依次發(fā)送給各節(jié)點(diǎn)：

public void start(Set<DiscoveryNode> faultyNodes) {
    logger.trace("publishing {} to {}", publishRequest, publicationTargets);
    publicationTargets.forEach(PublicationTarget::sendPublishRequest);
}
void sendPublishRequest() {
    Publication.this.sendPublishRequest(discoveryNode, publishRequest, new PublishResponseHandler());
}

復(fù)制

????????sendPublishRequest 內(nèi)部會(huì)分全量和增量兩種發(fā)送方式，在節(jié)點(diǎn)脫離之后重新加入或者有版本落后的情況下元數(shù)據(jù)會(huì)全量發(fā)送。至此 master 節(jié)點(diǎn)的發(fā)送流程就結(jié)束了，接下來就會(huì)在上面的 PublishResponseHandler 中等待多數(shù)節(jié)點(diǎn)響應(yīng)后發(fā)起 commit 請求。

6.2.2 數(shù)據(jù)節(jié)點(diǎn)接收 publish

????????實(shí)際處理 publish 的節(jié)點(diǎn)不止數(shù)據(jù)節(jié)點(diǎn)，還包括不帶數(shù)據(jù)屬性的所有節(jié)點(diǎn)，也包括當(dāng)前 master 節(jié)點(diǎn)本身。只是我們?yōu)榱朔奖忝枋觯詳?shù)據(jù)節(jié)點(diǎn)處理 publish 邏輯為主。數(shù)據(jù)節(jié)點(diǎn)接收 publish 請求的入口在 PublicationTransportHandler.handleIncomingPublishRequest()，接收的 ClusterState 分為全量和增量兩種，用一個(gè) boolean 區(qū)分，代碼片段：

private PublishWithJoinResponse handleIncomingPublishRequest(BytesTransportRequest request) throws IOException {
    StreamInput in = request.bytes().streamInput();
    try {
        ......
        // 全量或增量標(biāo)記
        if (in.readBoolean()) {
            final ClusterState incomingState;
            ......
            final PublishWithJoinResponse response = acceptState(incomingState);
            // 這里 lastSeenClusterState 的目的就是為了下次接收增量之后應(yīng)用 diff
            lastSeenClusterState.set(incomingState);
            return response;
        } else {
            final ClusterState lastSeen = lastSeenClusterState.get();
            if (lastSeen == null) {
                throw new IncompatibleClusterStateVersionException("have no local cluster state");
            } else {
                // 基于 lastSeen 應(yīng)用增量 diff 產(chǎn)生新的完整的 incomingState
                ClusterState incomingState;
                try {
                    final Diff<ClusterState> diff;
                    // Close stream early to release resources used by the de-compression as early as possible
                    try (StreamInput input = in) {
                        diff = ClusterState.readDiffFrom(input, lastSeen.nodes().getLocalNode());
                    }
                    incomingState = diff.apply(lastSeen); // might throw IncompatibleClusterStateVersionException
                } catch (Exception e) {
                    throw e;
                }
                ......
                final PublishWithJoinResponse response = acceptState(incomingState);
                // 本地變量 CAS 替換成全局
                lastSeenClusterState.compareAndSet(lastSeen, incomingState);
                return response;
            }
        }
    } finally {
        IOUtils.close(in);
    }
}

復(fù)制

????????序列化完畢后進(jìn)入本地 accept 階段。上述片段中 acceptState 函數(shù)內(nèi)部先會(huì)做元數(shù)據(jù)的各種校驗(yàn)，比如 term、clusterUUID 等最終更新 lastAcceptedState 并持久化，在元數(shù)據(jù)未 commit 之前元數(shù)據(jù)的 metaData 部分的 clusterUUIDCommitted 屬性為 false。元數(shù)據(jù)的標(biāo)識(shí)是通過 clusterUUID 來區(qū)分的，判斷一個(gè)集群的多個(gè)節(jié)點(diǎn)具備相同的元數(shù)據(jù)版本可根據(jù)此屬性，并結(jié)合 clusterUUIDCommitted 屬性確定元數(shù)據(jù)是已經(jīng) commit 的版本。處理接收請求代碼片段：

public PublishResponse handlePublishRequest(PublishRequest publishRequest) {
    final ClusterState clusterState = publishRequest.getAcceptedState();
    if (clusterState.term() != getCurrentTerm()) {
       ......
    }
    if (clusterState.term() == getLastAcceptedTerm() && clusterState.version() <= getLastAcceptedVersion()) {
       ......
    }
    // 更新 lastAcceptedState
    persistedState.setLastAcceptedState(clusterState);
    return new PublishResponse(clusterState.term(), clusterState.version());
}

復(fù)制

6.2.3 Master 節(jié)點(diǎn)發(fā)起 commit

????????在 master 節(jié)點(diǎn)發(fā)送 publish 請求給各個(gè)節(jié)點(diǎn)后，會(huì)在等待半數(shù)以上節(jié)點(diǎn)響應(yīng)才會(huì)進(jìn)入 commit 流程。代碼片段：

public Optional<ApplyCommitRequest> handlePublishResponse(DiscoveryNode sourceNode, PublishResponse publishResponse) {
    ......
    publishVotes.addVote(sourceNode);
    if (isPublishQuorum(publishVotes)) {
        return Optional.of(new ApplyCommitRequest(localNode, publishResponse.getTerm(), publishResponse.getVersion()));
    }
    return Optional.empty();
}

復(fù)制

一旦達(dá)到多數(shù)節(jié)點(diǎn)響應(yīng)后，master 發(fā)起 apply commit 請求：

@Override
protected void sendApplyCommit(
    DiscoveryNode destination,
    ApplyCommitRequest applyCommit,
    ActionListener<Empty> responseActionListener
) {
    transportService.sendRequest(
        destination,
        COMMIT_STATE_ACTION_NAME,
        applyCommit,
        COMMIT_STATE_REQUEST_OPTIONS,
        new ActionListenerResponseHandler<>(wrapWithMutex(responseActionListener), in -> Empty.INSTANCE, Names.CLUSTER_COORDINATION)
    );
}

復(fù)制

ApplyCommitRequest 請求比較簡單，僅包含 sourceNode、term、version 三個(gè)字段。

6.2.4 數(shù)據(jù)節(jié)點(diǎn)處理 commit

????????數(shù)據(jù)節(jié)點(diǎn)處理 commit 請求的入口在 Coordinator.java。一方面標(biāo)記接收的元數(shù)據(jù)為 commited 狀態(tài)，另外進(jìn)行元數(shù)據(jù)應(yīng)用。代碼片段：

private void handleApplyCommit(ApplyCommitRequest applyCommitRequest, ActionListener<Void> applyListener) {
    synchronized (mutex) {
        // 將我們之前接收到的 acceptedState 的元數(shù)據(jù)改為 commit 狀態(tài)
        coordinationState.get().handleCommit(applyCommitRequest);
        ......
        if (applyCommitRequest.getSourceNode().equals(getLocalNode())) {
            // master node applies the committed state at the end of the publication process, not here.
            applyListener.onResponse(null);
        } else {
            // 元數(shù)據(jù)接收完畢，進(jìn)入應(yīng)用環(huán)節(jié)
            clusterApplier.onNewClusterState(applyCommitRequest.toString(), () -> applierState, applyListener.map(r -> {
                onClusterStateApplied();
                return r;
            }));
        }
    }
}

復(fù)制

????????數(shù)據(jù)節(jié)點(diǎn) commit 元數(shù)據(jù)之后，元數(shù)據(jù)的發(fā)布流程就完畢了，之后數(shù)據(jù)節(jié)點(diǎn)進(jìn)入異步應(yīng)用環(huán)節(jié)。異步單線程處理，線程名稱是 clusterApplierService#updateTask 我們在很多 jstack 中看到該線程名稱即為數(shù)據(jù)節(jié)點(diǎn)的元數(shù)據(jù)應(yīng)用線程。

????????元數(shù)據(jù)應(yīng)用流程核心邏輯在 ClusterApplierService.java 的 applyChanges 函數(shù)中，主要是處理一些列元數(shù)據(jù)變更附屬的任務(wù)，例如創(chuàng)建、刪除索引、template 維護(hù)等，另外 master 節(jié)點(diǎn)上還會(huì)回調(diào)一些元數(shù)據(jù)變更完成后關(guān)聯(lián)的 listener。

private void applyChanges(ClusterState previousClusterState, ClusterState newClusterState, String source, Recorder stopWatch) {
    ......
    logger.debug("apply cluster state with version {}", newClusterState.version());
    // 進(jìn)行元數(shù)據(jù)應(yīng)用，會(huì)按優(yōu)先級分為 low/normal/high 來應(yīng)用，例如創(chuàng)建索引屬于 high
    callClusterStateAppliers(clusterChangedEvent, stopWatch);

    logger.debug("set locally applied cluster state to version {}", newClusterState.version());
    // 更新本地最新 commited 的 clusterState
    state.set(newClusterState);

    // 這里一般是 master 節(jié)點(diǎn)發(fā)起元數(shù)據(jù) commit 結(jié)束后，再回調(diào)相應(yīng)的 listener
    callClusterStateListeners(clusterChangedEvent, stopWatch);
}

復(fù)制

????????我們主要以創(chuàng)建索引來介紹元數(shù)據(jù)變更主流程，上述 callClusterStateAppliers apply 元數(shù)據(jù)的環(huán)節(jié)會(huì)進(jìn)到 IndicesClusterStateService.java applyClusterState 函數(shù)，該函數(shù)有一連串應(yīng)用最新 state 的流程：

@Override
public synchronized void applyClusterState(final ClusterChangedEvent event) {
    // 最新 commit 的元數(shù)據(jù)
    final ClusterState state = event.state();
    ......
    // 刪除索引、分片，清理磁盤分片目錄
    deleteIndices(event); // also deletes shards of deleted indices

    // 刪除索引、分片，只是清理內(nèi)存對象，主要是針對 Close/Open 操作
    removeIndicesAndShards(event); // also removes shards of removed indices

    // 更新索引 settings、mapping 等
    updateIndices(event); // can also fail shards, but these are then guaranteed to be in failedShardsCache

    // 創(chuàng)建索引、分片
    createIndicesAndUpdateShards(state);
}

復(fù)制

????????createIndicesAndUpdateShards 主要處理索引創(chuàng)建任務(wù)，根據(jù)接收到的元數(shù)據(jù)，對比內(nèi)存最新的索引列表，找出需要?jiǎng)?chuàng)建的索引列表進(jìn)行創(chuàng)建，同時(shí)創(chuàng)建缺失的分片信息：

private void createIndicesAndUpdateShards(final ClusterState state) {
    DiscoveryNodes nodes = state.nodes();
    // 通過節(jié)點(diǎn)到分片的映射，獲取屬于該節(jié)點(diǎn)的索引分片信息
    RoutingNode localRoutingNode = state.getRoutingNodes().node(nodes.getLocalNodeId());
  
    // 對比接收到的元數(shù)據(jù)和本節(jié)點(diǎn)內(nèi)存的索引列表，找出新增的需要?jiǎng)?chuàng)建的索引
    final Map<Index, List<ShardRouting>> indicesToCreate = new HashMap<>();
    for (ShardRouting shardRouting : localRoutingNode) {
        ShardId shardId = shardRouting.shardId();
        if (failedShardsCache.containsKey(shardId) == false) {
            final Index index = shardRouting.index();
            final var indexService = indicesService.indexService(index);
            if (indexService == null) {
                // 不存在的索引就是需要?jiǎng)?chuàng)建的
                indicesToCreate.computeIfAbsent(index, k -> new ArrayList<>()).add(shardRouting);
            } else {
                // 索引存在看看有沒需要?jiǎng)?chuàng)建或更新的分片
                createOrUpdateShard(state, nodes, routingTable, shardRouting, indexService);
            }
        }
    }

    // 過濾出來的待創(chuàng)建索引列表，遍歷依次創(chuàng)建
    for (Map.Entry<Index, List<ShardRouting>> entry : indicesToCreate.entrySet()) {
        final Index index = entry.getKey();
        final IndexMetadata indexMetadata = state.metadata().index(index);
        logger.debug("[{}] creating index", index);

        try {
            // 創(chuàng)建索引
            indicesService.createIndex(indexMetadata, buildInIndexListener, true);
        } catch (Exception e) {
            ......
        }
        // 創(chuàng)建索引對應(yīng)的本地分片
        for (ShardRouting shardRouting : entry.getValue()) {
            createOrUpdateShard(state, nodes, routingTable, shardRouting, indexService);
        }
    }
}

復(fù)制

????????至此，一個(gè)完整的元數(shù)據(jù)變更流程就介紹完了?？傮w來看，master 經(jīng)過兩階段提交元數(shù)據(jù)后，進(jìn)入元數(shù)據(jù)應(yīng)用流程，各個(gè)節(jié)點(diǎn)對比自己本地的信息和接收的元數(shù)據(jù)，根據(jù)差異處理相關(guān)流程。

7. 擴(kuò)展性優(yōu)化

????????前面我們介紹了 ES 元數(shù)據(jù)管理模型，接下來我們結(jié)合元數(shù)據(jù)管理模型看看 ES 分布式架構(gòu)存在的擴(kuò)展性瓶頸及優(yōu)化措施。

7.1 擴(kuò)展性瓶頸

????????社區(qū)版本建議控制整個(gè)集群分片數(shù)在 3 萬以下，節(jié)點(diǎn)數(shù)不超過100，超過之后在創(chuàng)建、刪除索引、維護(hù) mapping、template 等元數(shù)據(jù)變更操作時(shí)可能出現(xiàn)較嚴(yán)重的卡頓。例如 ES 在寫入觸發(fā)創(chuàng)建的場景，大批量 bulk 請求在索引創(chuàng)建完畢之前會(huì)堆積內(nèi)存，節(jié)點(diǎn)有被打垮的風(fēng)險(xiǎn)。其次，因?yàn)閱渭褐С值姆制瑪?shù)、節(jié)點(diǎn)數(shù)有限，導(dǎo)致用戶大規(guī)模數(shù)據(jù)場景下，需要建大量的小規(guī)格集群滿足業(yè)務(wù)需求，從而導(dǎo)致集群碎片化資源嚴(yán)重，整體 TCO 偏高。從前面元數(shù)據(jù)整體的變更流程中我們總結(jié)出如下主要的瓶頸：

• master 構(gòu)建新元數(shù)據(jù)，例如節(jié)點(diǎn)分片分配、均衡策略時(shí)，會(huì)基于 RoutingTable 全量構(gòu)建 RoutingNodes。
• master 節(jié)點(diǎn)元數(shù)據(jù)序列化，全量比對新舊元數(shù)據(jù)，構(gòu)建 diff 并序列化。
• data 節(jié)點(diǎn)元數(shù)據(jù) diff 推導(dǎo)，基于 RoutingTable 全量構(gòu)建 RoutingNodes，多次全量遍歷本節(jié)點(diǎn)分片和 diff 比對。
• 單個(gè)任務(wù)多次元數(shù)據(jù)變更，例如創(chuàng)建一個(gè)索引，會(huì)先創(chuàng)建主分片再創(chuàng)建從分片，導(dǎo)致單任務(wù)多次元數(shù)據(jù)變更。
• 路由全節(jié)點(diǎn)分發(fā)，單次元數(shù)據(jù)變更全節(jié)點(diǎn)分發(fā)，GC、網(wǎng)絡(luò)抖動(dòng)等長尾節(jié)點(diǎn)影響變更。
• 部分場景元數(shù)據(jù)同步落盤，HDD 場景性能影響嚴(yán)重。

???????總結(jié)來說，分片到節(jié)點(diǎn)的映射（RoutingTable）和節(jié)點(diǎn)到分片的映射（RoutingNodes）兩者之間的全量構(gòu)建，以及新舊元數(shù)據(jù)全量對比、全節(jié)點(diǎn)分發(fā)、同步落盤等是影響擴(kuò)展性的主要瓶頸。

7.2 優(yōu)化措施

????????結(jié)合前面分析的瓶頸點(diǎn)，優(yōu)化措施主要包括映射增量維護(hù)、收斂元數(shù)據(jù)變更范圍、重啟性能優(yōu)化等方向。

7.2.1 映射增量維護(hù)

????????RoutingTable 和 RoutingNodes 全量構(gòu)建的問題，在分片總數(shù)達(dá)到 5 萬以上時(shí)，多處這種相互構(gòu)建的性能瓶頸明顯，我們可以采用兩者增量維護(hù)的方式，避免全量相互構(gòu)建：

映射增量維護(hù)

????????元數(shù)據(jù)發(fā)送前序列化階段也不需要全量對比，因?yàn)槲覀兩厦嬉呀?jīng)維護(hù)了一個(gè) diff 的增量，可以省去一些全量對比的環(huán)節(jié)，直接將產(chǎn)生好的 diff 發(fā)送給數(shù)據(jù)節(jié)點(diǎn)即可。數(shù)據(jù)節(jié)點(diǎn)上的元數(shù)據(jù)推導(dǎo)也不需要再進(jìn)行全量對比，直接拿增量的 diff 進(jìn)行應(yīng)用。對于部分持久化存在同步刷盤的情況改為異步刷盤，緩解 HDD 盤場景的性能瓶頸。整體的優(yōu)化方向是流程增量、異步化：

元數(shù)據(jù)變更增量、異步化改造

7.2.2 收斂變更范圍

????????對于元數(shù)據(jù)發(fā)布全節(jié)點(diǎn)的瓶頸，在節(jié)點(diǎn)數(shù)多了例如數(shù)百個(gè)之后，容易因個(gè)別節(jié)點(diǎn)的抖動(dòng)受影響。我們可以控制元數(shù)據(jù)發(fā)布的范圍，只發(fā)生在 master 節(jié)點(diǎn)之間，索引、分片的創(chuàng)建采用定向節(jié)點(diǎn)下發(fā)的方式避免請求扇出嚴(yán)重，而數(shù)據(jù)節(jié)點(diǎn)可以通過學(xué)習(xí)的方式動(dòng)態(tài)構(gòu)建元數(shù)據(jù)。

收斂元數(shù)據(jù)變更范圍

7.2.3 重啟性能優(yōu)化

????????在集群的總分片數(shù)和節(jié)點(diǎn)數(shù)達(dá)到一定規(guī)模后，重啟恢復(fù)的性能會(huì)成為瓶頸，時(shí)間會(huì)比較長，也是我們需要主要優(yōu)化的方向。因?yàn)橹貑?huì)涉及到大量的分片分配、恢復(fù)，主要的優(yōu)化思路包括：

• 排序粗化：分片恢復(fù)有優(yōu)先級，按優(yōu)先級排序時(shí)，我們可以從分片維度優(yōu)化到索引維度。上面就是整體的擴(kuò)展性優(yōu)化的主要方向，還有其它方面的優(yōu)化，例如統(tǒng)計(jì)接口層面的優(yōu)化，主要思路是構(gòu)建緩存、低頻的方式提升接口性能。
• 批量 fetch：在恢復(fù)之前，master 會(huì)先獲取所有數(shù)據(jù)節(jié)點(diǎn)上分片最新的狀態(tài)，可以從單分片請求優(yōu)化為按節(jié)點(diǎn)批量獲取。
• 遍歷裁剪：在分片恢復(fù)過程中，每分配一批（并發(fā)恢復(fù)分片數(shù)控制）分片，都需要全量遍歷所有分片，遍歷過程中可以根據(jù)并發(fā)控制的分片數(shù)對遍歷進(jìn)行高效裁剪，去掉不需要分配的無用分片的遍歷過程。

7.2.4 優(yōu)化效果

????????在經(jīng)過一系列全方位擴(kuò)展性瓶頸優(yōu)化之后，我們將集群的分片數(shù)擴(kuò)展至百萬級，節(jié)點(diǎn)數(shù)擴(kuò)展至數(shù)百上千級。于此同時(shí)，我們也將部分熱點(diǎn)瓶頸優(yōu)化反饋給了社區(qū)。附部分已合并 PR：

• https://github.com/elastic/elasticsearch/pull/87723
• https://github.com/elastic/elasticsearch/pull/64753
• https://github.com/elastic/elasticsearch/pull/46520
• https://github.com/elastic/elasticsearch/pull/65045
• https://github.com/elastic/elasticsearch/pull/56870
• https://github.com/elastic/elasticsearch/pull/65172
• https://github.com/elastic/elasticsearch/pull/60564

8 總結(jié)

????????本文通過介紹 ES 分布式架構(gòu)，對比業(yè)界主流的分布式元數(shù)據(jù)管理模式，之后剖析了 ES 核心元數(shù)據(jù)管理模型。最后結(jié)合集群擴(kuò)展性的瓶頸介紹了騰訊在 ES 擴(kuò)展性方面所做的相關(guān)優(yōu)化，現(xiàn)階段重點(diǎn)解決了元數(shù)據(jù)的擴(kuò)展性，節(jié)點(diǎn)的擴(kuò)展性后續(xù)還需持續(xù)優(yōu)化以支持更大規(guī)模，目前騰訊云 ES 內(nèi)核已能滿足絕大部分用戶的擴(kuò)展性需求。文章來源地址http://www.zghlxwxcb.cn/news/detail-810793.html

參考文獻(xiàn)

• https://www.elastic.co/guide/en/elasticsearch/guide/index.html
• https://hadoop.apache.org/docs/r3.3.4/hadoop-project-dist/hadoop-hdfs/HdfsDesign.html
• https://cassandra.apache.org/_/cassandra-basics.html
• https://static.googleusercontent.com/media/research.google.com/zh-CN//archive/bigtable-osdi06.pdf

到了這里，關(guān)于Elasticsearch 分布式架構(gòu)剖析及擴(kuò)展性優(yōu)化的文章就介紹完了。如果您還想了解更多內(nèi)容，請?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！