Elasticsearch系列---分布式架構機制講解-有解無憂

概要

本篇主要介紹Elasticsearch的資料索引時的分片機制，集群發現機制，primary shard與replica shard是如何分工合作的，如何對集群擴容，以及集群的容錯機制，

分片機制

前面基本概念一節中，我們有提到建立索引時，會自動將資料拆分到多個分片（shard）中，默認數量是5，這個就是索引資料分片機制，我們在往Elasticsearch集群插入資料，并沒有關心過資料最終落地到哪個shard上，這個程序對客戶端來講是透明的，

document路由原理

document要存盤到Elasticsearch中，還要滿足后續搜索的需求，路由到分片位置的演算法肯定不能是隨機的，要不然搜索就沒法找了，路由的程序有一個公式：

shard = hash(routing) % number_of_primary_shards

routing值默認是document的ID值，也可以自行指定，先對routing資訊求hash值，然后將hash結果對primary_shard的數量求模，比如說primary_shard是5，那么結果肯定落在[0,4]區間內，這個結果值就是該document的分片位置，如示意圖所示：

document路由程序

這個求模公式間接的解釋了為什么了索引創建時指定了primary shard的值，后續就不讓改了，模數改了，之前路由的document再執行該公式時，值就可能跟改之前得到的值不一致，這樣document就找不到了，如示意圖所示：

修改shard值后路由失敗圖

集群發現機制

在同一個網路環境下，只要啟動一個Elasticsearch實體，并且cluster.name配置得一樣，這個Elasticsearch實體就會自動加入到集群當中，這個是如何實作的？

這個依賴于Elasticsearch的自動發現機制Zen，在elasticsearch.yml組態檔中，有一行
discovery.zen.ping.unicast.hosts: ["192.168.17.137"]
表示單播發現方式，當該Elasticsearch實體啟動時，會向192.168.17.137主機發送請求，并得到整個集群里所有節點的狀態，然后去聯系master節點，并加入集群，

摘抄了獲取配置資訊，注冊discovery請求的部分原始碼如下：
org.elasticsearch.discovery.zen.ZenDiscovery啟動時的構造器，會呼叫org.elasticsearch.discovery.zen.UnicastZenPing的構造器，其中UnicastZenPing的構造方式內會加載discovery.zen.ping.unicast.hosts配置項，并發送"internal:discovery/zen/unicast"請求（代碼有刪節）：

public UnicastZenPing(Settings settings, ThreadPool threadPool, TransportService transportService,
                          UnicastHostsProvider unicastHostsProvider, PingContextProvider contextProvider) {
        super(settings);

        final int concurrentConnects = DISCOVERY_ZEN_PING_UNICAST_CONCURRENT_CONNECTS_SETTING.get(settings);
        if (DISCOVERY_ZEN_PING_UNICAST_HOSTS_SETTING.exists(settings)) {
            configuredHosts = DISCOVERY_ZEN_PING_UNICAST_HOSTS_SETTING.get(settings);
            // we only limit to 1 addresses, makes no sense to ping 100 ports
            limitPortCounts = LIMIT_FOREIGN_PORTS_COUNT;
        } else {
            // if unicast hosts are not specified, fill with simple defaults on the local machine
            configuredHosts = transportService.getLocalAddresses();
            limitPortCounts = LIMIT_LOCAL_PORTS_COUNT;
        }
        resolveTimeout = DISCOVERY_ZEN_PING_UNICAST_HOSTS_RESOLVE_TIMEOUT.get(settings);


        transportService.registerRequestHandler(ACTION_NAME, ThreadPool.Names.SAME, UnicastPingRequest::new,
            new UnicastPingRequestHandler());
    }

shard&replica規則

一個index的資料，是拆分存盤在多個shard當中，我們可以在Elasticsearch的資料目錄里查看一下索引的存盤結構（Elasticsearch服務器上匯出的樹狀目錄結構）：

.
└── nodes
    └── 0
        ├── indices
        │   ├── 48G_CgE7TiWomlYsyQW1NQ #索引location的UUID
        │   │   ├── 0 #primary shard，從0-4共5個
        │   │   │   ├── index
        │   │   │   │   ├── segments_3
        │   │   │   │   └── write.lock
        │   │   │   ├── _state
        │   │   │   │   └── state-2.st
        │   │   │   └── translog
        │   │   │       ├── translog-2.ckp
        │   │   │       ├── translog-2.tlog
        │   │   │       ├── translog-3.ckp
        │   │   │       ├── translog-3.tlog
        │   │   │       ├── translog-4.tlog
        │   │   │       └── translog.ckp
        │   │   ├── 1
        │   │   │   ├── index
        │   │   │   │   ├── segments_3
        │   │   │   │   └── write.lock
        │   │   │   ├── _state
        │   │   │   │   └── state-2.st
        │   │   │   └── translog
        │   │   │       ├── translog-2.ckp
        │   │   │       ├── translog-2.tlog
        │   │   │       ├── translog-3.ckp
        │   │   │       ├── translog-3.tlog
        │   │   │       ├── translog-4.tlog
        │   │   │       └── translog.ckp
        │   │   ├── 2
        │   │   │   ├── index
        │   │   │   │   ├── _1.cfe
        │   │   │   │   ├── _1.cfs
        │   │   │   │   ├── _1.si
        │   │   │   │   ├── segments_7
        │   │   │   │   └── write.lock
        │   │   │   ├── _state
        │   │   │   │   └── state-2.st
        │   │   │   └── translog
        │   │   │       ├── translog-4.ckp
        │   │   │       ├── translog-4.tlog
        │   │   │       ├── translog-5.ckp
        │   │   │       ├── translog-5.tlog
        │   │   │       ├── translog-6.tlog
        │   │   │       └── translog.ckp
        │   │   ├── 3
        │   │   │   ├── index
        │   │   │   │   ├── _1.cfe
        │   │   │   │   ├── _1.cfs
        │   │   │   │   ├── _1.si
        │   │   │   │   ├── segments_7
        │   │   │   │   └── write.lock
        │   │   │   ├── _state
        │   │   │   │   └── state-2.st
        │   │   │   └── translog
        │   │   │       ├── translog-4.ckp
        │   │   │       ├── translog-4.tlog
        │   │   │       ├── translog-5.ckp
        │   │   │       ├── translog-5.tlog
        │   │   │       ├── translog-6.tlog
        │   │   │       └── translog.ckp
        │   │   ├── 4
        │   │   │   ├── index
        │   │   │   │   ├── _0.cfe
        │   │   │   │   ├── _0.cfs
        │   │   │   │   ├── _0.si
        │   │   │   │   ├── segments_5
        │   │   │   │   └── write.lock
        │   │   │   ├── _state
        │   │   │   │   └── state-2.st
        │   │   │   └── translog
        │   │   │       ├── translog-3.ckp
        │   │   │       ├── translog-3.tlog
        │   │   │       ├── translog-4.ckp
        │   │   │       ├── translog-4.tlog
        │   │   │       ├── translog-5.tlog
        │   │   │       └── translog.ckp
        │   │   └── _state
        │   │       └── state-16.st
        ├── node.lock
        └── _state
            ├── global-88.st
            └── node-22.st

如上目錄結構所示，展示了location索引（UUID為48G_CgE7TiWomlYsyQW1NQ）的存盤資訊，共5個primary shard，編號從0-4，

primary shard與replica shard，還有其他幾點特性：

shard是最小的存盤單元，像上面的0，1，2目錄，承載部分資料，
document是最小的資料單元，只能存在一個primary shard中以及對應的replica shard中（可能有多個），不會拆分存盤，也不會存在于多個primary shard里，
replica shard是primary shard的資料副本，冗余存盤，負責容錯，也可以承擔查詢請求，
primary shard不會和自己的replica shard放在一臺機器上，否則容錯機制就失效了，但是可以和別的replica shard混搭，
primary shard的數量在創建索引的時是多少就多少，后續不能改，但replica shard的數量可以隨時修改，

擴容機制

擴容分為垂直擴容和水平擴容兩種，垂直擴容指增加單臺服務器的CPU、記憶體大小，磁盤容量，簡單來講就是換更強大的服務器；水平擴容就是增加機器數量，通過集群化部署與分布式的技術手段，也能構建出強大的計算和存盤能力，

二者簡單對比：

垂直擴容：操作簡單，無需要更改集群方案，缺點就是貴，成本呈指數上升，并且單臺服務器瓶頸很明顯，
水平擴容：業務經常采用，因為更省錢，可以多非常多的普通服務器搭建，缺點是節點數越多，集群內節點之間通信會出現網路擁塞的問題，

Elastisearch非常適合用水平擴容方案，能勝任上百個節點，支撐PB級別的資料規模，并且擴容操作后，每增加新的節點會觸發索引分片的重新分配，

舉個例子，假定Elasticsearch有2個節點，primary shard設定為3，replica shard設定為1，這樣1個索引就有3個primary shard，3個replica shard，P表示primary shard，R表示replica shard，分布示例圖如下：

shard分布示例圖

當新加入一個node-3時，觸發node-1和node-2的shard進行重新分配，假定P0和R1兩個shard移到node-3當中，如圖所示：

shard rebalance示例圖

重分配完成后，此時集群的示例如下：

shard重新分布示例圖

最后補充兩點：

同一個index的primay shard和replica shard不能在同一個機器上，但不同index的primary shard和replica shard可以混搭，
負載均衡也不是完全平均的，有的多有的少，Elasticsearch會根據當前情況自動分配shard，

容錯機制

單node環境下的容錯

假定Elasticsearch集群只有一個node，primary shard設定為3，replica shard設定為1，這樣1個索引就應該有3個primary shard，3個replica shard，但primary shard不能與其replica shard放在一個node里，導致replica shard無法分配，這樣集群的status為yellow，示例圖如下：

單node的active shard圖

集群可以正常作業，一旦出現node宕機，資料全部丟失，并且集群不可用，

結論：單node環境容錯性為0.

2臺node環境下的容錯

primary shard與replica shard的設定與上文相同，此時Elasticsearch集群只有2個node，shard分布如下圖所示：

shard分布示例圖

如果其中一臺宕機，如node-2宕機，如圖所示：

模擬一臺node宕機

此時node-1節點的R2(replica shard)會升為P2(primary shard)，此時集群還能正常用，資料未丟失，

結論：雙node環境容錯性為1，

3臺node環境下的容錯

我們先按primary shard為3，replica shard為1進行容錯性計算，
此時每臺node存放2個shard，如果一臺宕機，此時另外2臺肯定還有完整的資料，如果兩臺宕機，剩下的那臺就只有2/3的資料，資料丟失1/3，容錯性為1臺，
如果是這樣設定，那3臺的容錯性和2臺的容錯性一樣，就存在資源浪費的情況，

那怎么樣提升容錯性呢？
把replica shard的值改成2，這樣每臺node存放3個shard，如下圖所示：

replica=2 shard分布示例圖

如果有2臺宕機，就剩下node-2，此時集群的資料還是完整的，replica會升成primary shard繼續提供服務，如下圖所示：

模擬兩臺node宕機

結論：3臺node環境容錯性最大可以是2，

擴容極限與最佳實踐

根據上面3個場景，我們可以知道，如果shard總數是6個（包含primary shard 和replica shard），那么node數量上限也為6，即每臺node存盤1個shard，這個資料即為擴容極限，如果要突破極限，可以通過增大replica的值來實作，這樣有更多的replica shard去分擔查詢請求，占用更多的節點，整個集群的CPU、IO、Memory資源更多，整體吞吐量也越高，

當然這個replica也不是越大越好，冗余存盤占用磁盤資源，replica越大，集群內有效資料的磁盤利用率就越低，以3臺node為例，想要達到容錯性，磁盤利用率的最佳值，replica=2是最適宜的，

實際生產中，可以根據資料量，并發數等實際需求，在創建索引時合理設定primary shard的數量，后期優化時，再調整replica shard的值，這個需要反復驗證，不斷的演算調整，最終讓生產Elasticsearch集群的吞吐量達到一個最佳值，

容錯程序與選舉機制

Elasticsearch集群中，所有的node都是對等的角色，所有的node都能接收請求，并且能自動轉請求到相應的節點上（資料路由），最后能將其他節點處理的資料進行回應收集，回傳給客戶端，在集群中，也存在一個master節點，它的職責多一些，需要管理與維護集群的元資料，索引的創建與洗掉和節點的增加和洗掉，它都會收到相應的請求，然后進行相應的資料維護，master node在承擔索引、搜索請求時，與其他node一起分攤，并不承擔所有的請求，因而不存在單點故障這個問題，

我們假設一下集群有3臺node，其中node-1宕機的程序，如果node-1是master node，關鍵步驟如下：

3臺node replica=1 shard分布示例圖

丟失了3個shard，由于P1丟失，cluster.status瞬間狀態變成red，
重新進行master選舉，自動選另一個node作為master，
新的master將丟失了P1對應的R1（在node-3上面）提升為primary shard ，現全部primary shard active，但是P1，P2的replica shard無法啟動，cluster.status變成yellow，
重啟故障的node-1節點，新的master會將缺失的副本都copy一份到node-1上，node-1會使用之間已有的資料，并且同步一下宕機期間的資料修改，此時所有的shard全部active狀態，cluster.status重新變成green，

小結

本篇針對Elasticsearch的一些內部原理進行了簡單的介紹，這些原理針對Elasticsearch的使用者是透明的，為了增加可閱讀性，自行增加一些講解的原理圖，若有不詳盡之處或錯誤之處請指正，謝謝，

專注Java高并發、分布式架構，更多技術干貨分享與心得，請關注公眾號：Java架構社區

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

標籤：其他

上一篇：Elasticsearch系列---常見搜索方式與聚合分析

下一篇：docker打包python應用