一、高可用的由來
1.1 為何需要Replication
在Kafka在0.8以前的版本中眯停，是沒有Replication的，一旦某一個Broker宕機，則其上所有的Partition數(shù)據(jù)都不可被消費备恤，這與Kafka數(shù)據(jù)持久性及Delivery Guarantee的設(shè)計目標相悖。同時Producer都不能再將數(shù)據(jù)存于這些Partition中歉闰。
如果Producer使用同步模式則Producer會在嘗試重新發(fā)送message.send.max.retries（默認值為3）次后拋出Exception沸版，用戶可以選擇停止發(fā)送后續(xù)數(shù)據(jù)也可選擇繼續(xù)選擇發(fā)送。而前者會造成數(shù)據(jù)的阻塞楞件，后者會造成本應(yīng)發(fā)往該Broker的數(shù)據(jù)的丟失衫生。
如果Producer使用異步模式，則Producer會嘗試重新發(fā)送message.send.max.retries（默認值為3）次后記錄該異常并繼續(xù)發(fā)送后續(xù)數(shù)據(jù)土浸，這會造成數(shù)據(jù)丟失并且用戶只能通過日志發(fā)現(xiàn)該問題罪针。同時，Kafka的Producer并未對異步模式提供callback接口黄伊。
由此可見泪酱，在沒有Replication的情況下，一旦某機器宕機或者某個Broker停止工作則會造成整個系統(tǒng)的可用性降低还最。隨著集群規(guī)模的增加墓阀，整個集群中出現(xiàn)該類異常的幾率大大增加，因此對于生產(chǎn)系統(tǒng)而言Replication機制的引入非常重要拓轻。
1.2 Leader Election
引入Replication之后岂津，同一個Partition可能會有多個Replica，而這時需要在這些Replication之間選出一個Leader悦即，Producer和Consumer只與這個Leader交互吮成，其它Replica作為Follower從Leader中復(fù)制數(shù)據(jù)。
因為需要保證同一個Partition的多個Replica之間的數(shù)據(jù)一致性（其中一個宕機后其它Replica必須要能繼續(xù)服務(wù)并且即不能造成數(shù)據(jù)重復(fù)也不能造成數(shù)據(jù)丟失）辜梳。如果沒有一個Leader粱甫，所有Replica都可同時讀/寫數(shù)據(jù)，那就需要保證多個Replica之間互相（N×N條通路）同步數(shù)據(jù)作瞄，數(shù)據(jù)的一致性和有序性非常難保證茶宵，大大增加了Replication實現(xiàn)的復(fù)雜性，同時也增加了出現(xiàn)異常的幾率宗挥。而引入Leader后乌庶，只有Leader負責(zé)數(shù)據(jù)讀寫种蝶，F(xiàn)ollower只向Leader順序Fetch數(shù)據(jù)（N條通路），系統(tǒng)更加簡單且高效瞒大。
二螃征、Kafka HA設(shè)計解析
2.1 如何將所有Replica均勻分布到整個集群
為了更好的做負載均衡，Kafka盡量將所有的Partition均勻分配到整個集群上透敌。一個典型的部署方式是一個Topic的Partition數(shù)量大于Broker的數(shù)量盯滚。同時為了提高Kafka的容錯能力，也需要將同一個Partition的Replica盡量分散到不同的機器酗电。實際上魄藕，如果所有的Replica都在同一個Broker上，那一旦該Broker宕機撵术，該Partition的所有Replica都無法工作背率，也就達不到HA的效果。同時嫩与，如果某個Broker宕機了退渗，需要保證它上面的負載可以被均勻的分配到其它幸存的所有Broker上。
Kafka分配Replica的算法如下：
1.將所有Broker（假設(shè)共n個Broker）和待分配的Partition排序
2.將第i個Partition分配到第（i mod n）個Broker上
3.將第i個Partition的第j個Replica分配到第（(i + j) mode n）個Broker上
2.2 Data Replication（副本策略）
Kafka的高可靠性的保障來源于其健壯的副本（replication）策略蕴纳。
2.2.1 消息傳遞同步策略
Producer在發(fā)布消息到某個Partition時，先通過ZooKeeper找到該Partition的Leader个粱，然后無論該Topic的Replication Factor為多少古毛，Producer只將該消息發(fā)送到該Partition的Leader。Leader會將該消息寫入其本地Log都许。每個Follower都從Leader pull數(shù)據(jù)稻薇。這種方式上，F(xiàn)ollower存儲的數(shù)據(jù)順序與Leader保持一致胶征。Follower在收到該消息并寫入其Log后塞椎，向Leader發(fā)送ACK。一旦Leader收到了ISR中的所有Replica的ACK睛低，該消息就被認為已經(jīng)commit了案狠，Leader將增加HW并且向Producer發(fā)送ACK。
為了提高性能钱雷，每個Follower在接收到數(shù)據(jù)后就立馬向Leader發(fā)送ACK骂铁，而非等到數(shù)據(jù)寫入Log中。因此罩抗，對于已經(jīng)commit的消息拉庵，Kafka只能保證它被存于多個Replica的內(nèi)存中，而不能保證它們被持久化到磁盤中套蒂，也就不能完全保證異常發(fā)生后該條消息一定能被Consumer消費钞支。
Consumer讀消息也是從Leader讀取茫蛹，只有被commit過的消息才會暴露給Consumer。
Kafka Replication的數(shù)據(jù)流如下圖所示：
2.2.2 ACK前需要保證有多少個備份
對于Kafka而言烁挟，定義一個Broker是否“活著”包含兩個條件：
一是它必須維護與ZooKeeper的session（這個通過ZooKeeper的Heartbeat機制來實現(xiàn)）婴洼。
二是Follower必須能夠及時將Leader的消息復(fù)制過來，不能“落后太多”信夫。
Leader會跟蹤與其保持同步的Replica列表窃蹋，該列表稱為ISR（即in-sync Replica）。如果一個Follower宕機静稻，或者落后太多警没，Leader將把它從ISR中移除。這里所描述的“落后太多”指Follower復(fù)制的消息落后于Leader后的條數(shù)超過預(yù)定值（該值可在$KAFKA_HOME/config/server.properties中通過replica.lag.max.messages配置振湾，其默認值是4000）或者Follower超過一定時間（該值可在$KAFKA_HOME/config/server.properties中通過replica.lag.time.max.ms來配置杀迹，其默認值是10000）未向Leader發(fā)送fetch請求。
Kafka的復(fù)制機制既不是完全的同步復(fù)制押搪，也不是單純的異步復(fù)制树酪。事實上，完全同步復(fù)制要求所有能工作的Follower都復(fù)制完大州，這條消息才會被認為commit续语，這種復(fù)制方式極大的影響了吞吐率（高吞吐率是Kafka非常重要的一個特性）。而異步復(fù)制方式下厦画，F(xiàn)ollower異步的從Leader復(fù)制數(shù)據(jù)疮茄，數(shù)據(jù)只要被Leader寫入log就被認為已經(jīng)commit，這種情況下如果Follower都復(fù)制完都落后于Leader根暑，而如果Leader突然宕機力试，則會丟失數(shù)據(jù)。而Kafka的這種使用ISR的方式則很好的均衡了確保數(shù)據(jù)不丟失以及吞吐率排嫌。Follower可以批量的從Leader復(fù)制數(shù)據(jù)畸裳，這樣極大的提高復(fù)制性能（批量寫磁盤），極大減少了Follower與Leader的差距淳地。
需要說明的是怖糊，Kafka只解決fail/recover，不處理“Byzantine”（“拜占庭”）問題颇象。一條消息只有被ISR里的所有Follower都從Leader復(fù)制過去才會被認為已提交蓬抄。這樣就避免了部分數(shù)據(jù)被寫進了Leader，還沒來得及被任何Follower復(fù)制就宕機了夯到，而造成數(shù)據(jù)丟失（Consumer無法消費這些數(shù)據(jù)）嚷缭。而對于Producer而言，它可以選擇是否等待消息commit，這可以通過request.required.acks來設(shè)置阅爽。這種機制確保了只要ISR有一個或以上的Follower路幸，一條被commit的消息就不會丟失。
2.2.3 Leader Election算法
Leader選舉本質(zhì)上是一個分布式鎖付翁，有兩種方式實現(xiàn)基于ZooKeeper的分布式鎖：
節(jié)點名稱唯一性：多個客戶端創(chuàng)建一個節(jié)點简肴，只有成功創(chuàng)建節(jié)點的客戶端才能獲得鎖
臨時順序節(jié)點：所有客戶端在某個目錄下創(chuàng)建自己的臨時順序節(jié)點，只有序號最小的才獲得鎖
一種非常常用的選舉leader的方式是“Majority Vote”（“少數(shù)服從多數(shù)”）百侧，但Kafka并未采用這種方式砰识。這種模式下，如果我們有2f+1個Replica（包含Leader和Follower）佣渴，那在commit之前必須保證有f+1個Replica復(fù)制完消息辫狼，為了保證正確選出新的Leader，fail的Replica不能超過f個辛润。因為在剩下的任意f+1個Replica里膨处，至少有一個Replica包含有最新的所有消息。這種方式有個很大的優(yōu)勢砂竖，系統(tǒng)的latency只取決于最快的幾個Broker真椿，而非最慢那個。Majority Vote也有一些劣勢乎澄，為了保證Leader Election的正常進行突硝，它所能容忍的fail的follower個數(shù)比較少。如果要容忍1個follower掛掉置济，必須要有3個以上的Replica解恰，如果要容忍2個Follower掛掉，必須要有5個以上的Replica舟肉。也就是說，在生產(chǎn)環(huán)境下為了保證較高的容錯程度查库，必須要有大量的Replica路媚，而大量的Replica又會在大數(shù)據(jù)量下導(dǎo)致性能的急劇下降。這就是這種算法更多用在ZooKeeper這種共享集群配置的系統(tǒng)中而很少在需要存儲大量數(shù)據(jù)的系統(tǒng)中使用的原因樊销。例如HDFS的HA Feature是基于majority-vote-based journal整慎，但是它的數(shù)據(jù)存儲并沒有使用這種方式。
Kafka在ZooKeeper中動態(tài)維護了一個ISR（in-sync replicas）围苫，這個ISR里的所有Replica都跟上了leader裤园，只有ISR里的成員才有被選為Leader的可能。在這種模式下剂府，對于f+1個Replica拧揽，一個Partition能在保證不丟失已經(jīng)commit的消息的前提下容忍f個Replica的失敗。在大多數(shù)使用場景中，這種模式是非常有利的淤袜。事實上痒谴，為了容忍f個Replica的失敗，Majority Vote和ISR在commit前需要等待的Replica數(shù)量是一樣的铡羡，但是ISR需要的總的Replica的個數(shù)幾乎是Majority Vote的一半积蔚。
雖然Majority Vote與ISR相比有不需等待最慢的Broker這一優(yōu)勢，但是Kafka作者認為Kafka可以通過Producer選擇是否被commit阻塞來改善這一問題烦周，并且節(jié)省下來的Replica和磁盤使得ISR模式仍然值得尽爆。
2.2.4 如何處理所有Replica都不工作
在ISR中至少有一個follower時，Kafka可以確保已經(jīng)commit的數(shù)據(jù)不丟失读慎，但如果某個Partition的所有Replica都宕機了漱贱，就無法保證數(shù)據(jù)不丟失了。這種情況下有兩種可行的方案：
1.等待ISR中的任一個Replica“活”過來贪壳，并且選它作為Leader
2.選擇第一個“活”過來的Replica（不一定是ISR中的）作為Leader
這就需要在可用性和一致性當中作出一個簡單的折衷饱亿。如果一定要等待ISR中的Replica“活”過來，那不可用的時間就可能會相對較長闰靴。而且如果ISR中的所有Replica都無法“活”過來了彪笼，或者數(shù)據(jù)都丟失了，這個Partition將永遠不可用蚂且。選擇第一個“活”過來的Replica作為Leader配猫，而這個Replica不是ISR中的Replica，那即使它并不保證已經(jīng)包含了所有已commit的消息杏死，它也會成為Leader而作為consumer的數(shù)據(jù)源（前文有說明泵肄，所有讀寫都由Leader完成）。Kafka0.8.*使用了第二種方式淑翼。根據(jù)Kafka的文檔腐巢，在以后的版本中，Kafka支持用戶通過配置選擇這兩種方式中的一種玄括，從而根據(jù)不同的使用場景選擇高可用性還是強一致性冯丙。
2.2.5 選舉Leader
最簡單最直觀的方案是，所有Follower都在ZooKeeper上設(shè)置一個Watch遭京，一旦Leader宕機胃惜，其對應(yīng)的ephemeral znode會自動刪除，此時所有Follower都嘗試創(chuàng)建該節(jié)點哪雕，而創(chuàng)建成功者（ZooKeeper保證只有一個能創(chuàng)建成功）即是新的Leader船殉，其它Replica即為Follower。
但是該方法會有3個問題：
1.split-brain 這是由ZooKeeper的特性引起的斯嚎，雖然ZooKeeper能保證所有Watch按順序觸發(fā)利虫，但并不能保證同一時刻所有Replica“看”到的狀態(tài)是一樣的挨厚，這就可能造成不同Replica的響應(yīng)不一致
2.herd effect 如果宕機的那個Broker上的Partition比較多，會造成多個Watch被觸發(fā)列吼，造成集群內(nèi)大量的調(diào)整
3.ZooKeeper負載過重 每個Replica都要為此在ZooKeeper上注冊一個Watch幽崩，當集群規(guī)模增加到幾千個Partition時ZooKeeper負載會過重。
Kafka 0.8.*的Leader Election方案解決了上述問題寞钥，它在所有broker中選出一個controller慌申，所有Partition的Leader選舉都由controller決定。controller會將Leader的改變直接通過RPC的方式（比ZooKeeper Queue的方式更高效）通知需為為此作為響應(yīng)的Broker理郑。同時controller也負責(zé)增刪Topic以及Replica的重新分配蹄溉。
三、HA相關(guān)ZooKeeper結(jié)構(gòu)
3.1 admin
該目錄下znode只有在有相關(guān)操作時才會存在您炉，操作結(jié)束時會將其刪除
/admin/reassign_partitions用于將一些Partition分配到不同的broker集合上柒爵。對于每個待重新分配的Partition，Kafka會在該znode上存儲其所有的Replica和相應(yīng)的Broker id赚爵。該znode由管理進程創(chuàng)建并且一旦重新分配成功它將會被自動移除棉胀。
3.2 broker
即/brokers/ids/[brokerId]）存儲“活著”的broker信息。
topic注冊信息（/brokers/topics/[topic]）冀膝，存儲該topic的所有partition的所有replica所在的broker id唁奢，第一個replica即為preferred replica，對一個給定的partition窝剖，它在同一個broker上最多只有一個replica,因此broker id可作為replica id麻掸。
3.3 controller
/controller -> int (broker id of the controller)存儲當前controller的信息
/controller_epoch -> int (epoch)直接以整數(shù)形式存儲controller epoch，而非像其它znode一樣以JSON字符串形式存儲赐纱。
四脊奋、producer發(fā)布消息
4.1 寫入方式
producer 采用 push 模式將消息發(fā)布到 broker，每條消息都被 append 到 patition 中疙描，屬于順序?qū)懘疟P（順序?qū)懘疟P效率比隨機寫內(nèi)存要高诚隙，保障 kafka 吞吐率）。
4.2 消息路由
producer 發(fā)送消息到 broker 時起胰，會根據(jù)分區(qū)算法選擇將其存儲到哪一個 partition久又。其路由機制為：
1、 指定了 patition待错，則直接使用籽孙；
2烈评、 未指定 patition 但指定 key火俄，通過對 key 的 value 進行hash 選出一個 patition
3、 patition 和 key 都未指定讲冠，使用輪詢選出一個 patition瓜客。
4.3 寫入流程
producer 寫入消息序列圖如下所示：
流程說明：
1、 producer 先從 zookeeper 的 "/brokers/.../state" 節(jié)點找到該 partition 的 leader
2、 producer 將消息發(fā)送給該 leader
3谱仪、 leader 將消息寫入本地 log
4玻熙、 followers 從 leader pull 消息，寫入本地 log 后 leader 發(fā)送 ACK
5疯攒、 leader 收到所有 ISR 中的 replica 的 ACK 后嗦随，增加 HW（high watermark，最后 commit 的 offset） 并向 producer 發(fā)送 ACK
五敬尺、broker保存消息
5.1 存儲方式
物理上把 topic 分成一個或多個 patition（對應(yīng) server.properties 中的 num.partitions=3 配置）枚尼，每個 patition 物理上對應(yīng)一個文件夾（該文件夾存儲該 patition 的所有消息和索引文件），如下：
5.2 存儲策略
無論消息是否被消費砂吞，kafka 都會保留所有消息署恍。有兩種策略可以刪除舊數(shù)據(jù)：
1、 基于時間：log.retention.hours=168
2蜻直、 基于大卸⒅省：log.retention.bytes=1073741824
六、Topic的創(chuàng)建和刪除
6.1 創(chuàng)建topic
創(chuàng)建 topic 的序列圖如下所示：
流程說明：
1概而、 controller 在 ZooKeeper 的 /brokers/topics 節(jié)點上注冊 watcher呼巷，當 topic 被創(chuàng)建，則 controller 會通過 watch 得到該 topic 的 partition/replica 分配到腥。
2朵逝、 controller從 /brokers/ids 讀取當前所有可用的 broker 列表，對于 set_p 中的每一個 partition：
? ? 2.1乡范、 從分配給該 partition 的所有 replica（稱為AR）中任選一個可用的 broker 作為新的 leader配名，并將AR設(shè)置為新的 ISR
? ? 2.2、 將新的 leader 和 ISR 寫入 /brokers/topics/[topic]/partitions/[partition]/state
3晋辆、 controller 通過 RPC 向相關(guān)的 broker 發(fā)送 LeaderAndISRRequest渠脉。
6.2 刪除topic
刪除 topic 的序列圖如下所示：
流程說明：
1、 controller 在 zooKeeper 的 /brokers/topics 節(jié)點上注冊 watcher瓶佳，當 topic 被刪除芋膘，則 controller 會通過 watch 得到該 topic 的 partition/replica 分配。
2霸饲、 若 delete.topic.enable=false为朋，結(jié)束；否則 controller 注冊在 /admin/delete_topics 上的 watch 被 fire厚脉，controller 通過回調(diào)向?qū)?yīng)的 broker 發(fā)送 StopReplicaRequest习寸。
七、broker failover
kafka broker failover 序列圖如下所示：
流程說明：
1傻工、 controller 在 zookeeper 的 /brokers/ids/[brokerId] 節(jié)點注冊 Watcher霞溪，當 broker 宕機時 zookeeper 會 fire watch
2孵滞、 controller 從 /brokers/ids 節(jié)點讀取可用broker
3、 controller決定set_p鸯匹，該集合包含宕機 broker 上的所有 partition
4坊饶、 對 set_p 中的每一個 partition
? ? 4.1、 從/brokers/topics/[topic]/partitions/[partition]/state 節(jié)點讀取 ISR
? ? 4.2殴蓬、 決定新 leader
? ? 4.3匿级、 將新 leader、ISR染厅、controller_epoch 和 leader_epoch 等信息寫入 state 節(jié)點
5根蟹、 通過 RPC 向相關(guān) broker 發(fā)送 leaderAndISRRequest 命令
八、controller failover
當 controller 宕機時會觸發(fā) controller failover糟秘。每個 broker 都會在 zookeeper 的 "/controller" 節(jié)點注冊 watcher简逮，當 controller 宕機時 zookeeper 中的臨時節(jié)點消失，所有存活的 broker 收到 fire 的通知尿赚，每個 broker 都嘗試創(chuàng)建新的 controller path散庶，只有一個競選成功并當選為 controller。
當新的 controller 當選時凌净，會觸發(fā) KafkaController.onControllerFailover 方法悲龟，在該方法中完成如下操作：
1、 讀取并增加 Controller Epoch冰寻。
2须教、 在 reassignedPartitions Patch(/admin/reassign_partitions) 上注冊 watcher。
3斩芭、 在 preferredReplicaElection Path(/admin/preferred_replica_election) 上注冊 watcher轻腺。
4、 通過 partitionStateMachine 在 broker Topics Patch(/brokers/topics) 上注冊 watcher划乖。
5贬养、 若 delete.topic.enable=true（默認值是 false），則 partitionStateMachine 在 Delete Topic Patch(/admin/delete_topics) 上注冊 watcher琴庵。
6误算、 通過 replicaStateMachine在 Broker Ids Patch(/brokers/ids)上注冊Watch。
7迷殿、 初始化 ControllerContext 對象儿礼，設(shè)置當前所有 topic，“活”著的 broker 列表庆寺，所有 partition 的 leader 及 ISR等蚊夫。
8、 啟動 replicaStateMachine 和 partitionStateMachine止邮。
9这橙、 將 brokerState 狀態(tài)設(shè)置為 RunningAsController。
10导披、 將每個 partition 的 Leadership 信息發(fā)送給所有“活”著的 broker屈扎。
11、 若 auto.leader.rebalance.enable=true（默認值是true）撩匕，則啟動 partition-rebalance 線程鹰晨。
12、 若 delete.topic.enable=true 且Delete Topic Patch(/admin/delete_topics)中有值止毕，則刪除相應(yīng)的Topic模蜡。
