實(shí)時(shí)計(jì)算之kafka的高可用保證

kakfa集群介紹01kakfa服務(wù)的故障與恢復(fù)02kafka的HA機(jī)制03CONTENTSkafka集群簡單介紹kafka集群有4個(gè)kakfa集群cdn集群：40臺(tái)user-action集群：30臺(tái)性能測試集群：10臺(tái)cdn老集群：60臺(tái)（即將淘汰）kafka服務(wù)的不穩(wěn)定因素

磁盤損壞 會(huì)造成kafka服務(wù)異常退出。服務(wù)器故障 kafka服務(wù)退出；兩臺(tái)以上故障：kafka數(shù)據(jù)丟失等問題。zookeeper服務(wù) 之前所有的kafka和共用一套zookeeper服務(wù)，現(xiàn)已經(jīng)分拆。kafka服務(wù)器進(jìn)程監(jiān)控目前的監(jiān)控方式：使用crontab定時(shí)掃描監(jiān)測kafka服務(wù)進(jìn)程， 如kafka服務(wù)不存在，自動(dòng)拉起，并短信報(bào)警給相關(guān)人員。缺點(diǎn)：時(shí)效性不高。其他方式：daemontools監(jiān)控。KAFKA磁盤故障自動(dòng)恢復(fù)由于kafka服務(wù)器磁盤開銷很大，隨著集群規(guī)模的增大，出現(xiàn)磁盤故障的可能性大大提高，為了減少維護(hù)的成本，我們開發(fā)了磁盤故障的自動(dòng)處理程序。通過mysql來管理損壞磁盤；自動(dòng)剔除故障磁盤。自動(dòng)恢復(fù)磁盤服務(wù)器運(yùn)行日志目錄重新選舉。磁盤故障報(bào)警通知。kafka備份池kafka備份池：安裝了3臺(tái)kafka服務(wù)，平時(shí)不啟動(dòng)，線上kafka服務(wù)器故障時(shí)，從備份池中取出一臺(tái)備份服務(wù)器接替，同時(shí)通知IT及時(shí)維修故障的服務(wù)器。由于目前的kafka的topic副本數(shù)為2；當(dāng)kafka服務(wù)器故障超過兩臺(tái)時(shí)，就會(huì)造成數(shù)據(jù)丟失；當(dāng)kafka服務(wù)器故障時(shí)，可根據(jù)需要啟動(dòng)備份服務(wù)器。kafka備份機(jī)策略可行性測試檢驗(yàn)備份機(jī)策略是否可行，同時(shí)為了檢驗(yàn)服務(wù)器故障對(duì)服務(wù)的影響，檢驗(yàn)kafka備份機(jī)切換對(duì)集群和服務(wù)的影響，做了如下測試。測試方案：step1:部署3臺(tái)測試集群的kafka集群備份機(jī)。step2:創(chuàng)建一個(gè)測試的topic，只有一個(gè)分區(qū)，備份數(shù)為2的test_bak_kafkastep3:編寫一個(gè)發(fā)送消息的生產(chǎn)者程序，不停的發(fā)送消息，啟動(dòng)一個(gè)消費(fèi)者，實(shí)時(shí)消費(fèi)消息。step4:關(guān)閉一個(gè)正在寫入消息的broker的主節(jié)點(diǎn)stop5:啟動(dòng)被關(guān)閉的節(jié)點(diǎn)。step6:關(guān)閉主節(jié)點(diǎn)，啟動(dòng)備份節(jié)點(diǎn)

kafka的zk元數(shù)據(jù)信息介紹關(guān)閉topic分區(qū)的leader節(jié)點(diǎn)

現(xiàn)象：生產(chǎn)者寫消息的生產(chǎn)者會(huì)報(bào)如下異常退去，不可恢復(fù)。上層應(yīng)用需要考慮重試機(jī)制消費(fèi)者：不受影響。kafka服務(wù)器：topic的備份節(jié)點(diǎn)切換成了leader當(dāng)異常的broker節(jié)點(diǎn)退出時(shí),leader也會(huì)重新選舉，而異常節(jié)點(diǎn)加入時(shí)，不會(huì)重新選舉。如果故障退出的為follower節(jié)點(diǎn)，則生產(chǎn)者和消費(fèi)者都不會(huì)受到影響。啟動(dòng)備份節(jié)點(diǎn)步驟操作結(jié)果stop1：修改備份節(jié)點(diǎn)的borker.id為故障節(jié)點(diǎn)ID需提前配置備份節(jié)點(diǎn)到/etc/hoststop2：啟動(dòng)備份節(jié)點(diǎn)正常啟動(dòng)stop3：觀察zk中brokerid的變化zk中/brokers/ids的出現(xiàn)brokerid的節(jié)點(diǎn)stop4：觀察生產(chǎn)者和消費(fèi)者的狀態(tài)沒有影響stop5：觀察備份節(jié)點(diǎn)的數(shù)據(jù)同步狀態(tài)數(shù)據(jù)沒有同步過來，等待一段時(shí)間后，可以恢復(fù)同步。數(shù)據(jù)同步效率：1個(gè)磁盤：16分鐘53G的數(shù)據(jù)。54M/stopic的zk的狀態(tài)變化在kafka節(jié)點(diǎn)異常，備份節(jié)點(diǎn)恢復(fù)過程中，topic的zk狀態(tài)變化如下：關(guān)閉主節(jié)點(diǎn)：kafka的leader發(fā)生切換啟動(dòng)備份節(jié)點(diǎn)：正常狀態(tài)：一臺(tái)broker故障后：備份broker恢復(fù)后：kafka的LeaderElectionkafka的partition的leader算法沒有使用常用的“MajorityVote”；而是采用在ZK中動(dòng)態(tài)維護(hù)了一個(gè)ISR，這個(gè)ISR里的所有Replica都需要跟上了leader，只有ISR里的成員才有被選為Leader的可能。優(yōu)點(diǎn)：N+1個(gè)Replica，容忍N(yùn)個(gè)replica失敗容忍N(yùn)個(gè)Replica的失敗，比“MajorityVote”少近一半的replica缺點(diǎn)：需要等待最慢的Broker，但是可以通過Producer選擇是否被commit阻塞來改善（request.required.acks）。kafka的LeaderElection實(shí)現(xiàn)：kafka通過Controller節(jié)點(diǎn)來選舉各個(gè)partition的leader。kafka的LeaderElection1.controller在ZK的/brokers/ids節(jié)點(diǎn)上注冊Watch2.broker宕機(jī)，ZK會(huì)fireController注冊的Watch會(huì)觸發(fā)controller。3.Controller重新設(shè)置partition信息（set_p）。4.從/brokers/topics/[topic]/partitions/[partition]/state讀取該P(yáng)artition當(dāng)前的ISR。5.決定該P(yáng)artition的新Leader如果ISR列表中有Replica有幸存，選擇其中一個(gè)為新Leader，新ISR為當(dāng)前ISR；如果ISR列表中的Replica都當(dāng)宕掉，選擇任意一個(gè)幸存的Replica為新Leader，以及ISR（會(huì)有數(shù)據(jù)丟失的可能）；如果該P(yáng)artition的所有Replica都宕機(jī)了，則將新的Leader設(shè)置為-1；6.Controlle直接通過RPC向相關(guān)Broker發(fā)送LeaderAndISRRequest命令。如何處理所有Replica都不工作如果某個(gè)Partition的所有Replica都宕機(jī)了，就無法保證數(shù)據(jù)不丟失了。這種情況下有兩種可行的方案：1.等待ISR中的任一個(gè)Replica“活”過來，并且選它作為Leader2.選擇第一個(gè)“活”過來的Replica（不一定是ISR中的）作為Leader?？捎眯院鸵恢滦援?dāng)中一個(gè)簡單的折衷，kafka使用的第二種方法。kafka的DataReplication消息生產(chǎn)者的消息發(fā)送機(jī)制：同步發(fā)送:Producer會(huì)在嘗試重新發(fā)送message.send.max.retries次后拋出Exception異步發(fā)送：Producer會(huì)嘗試重新發(fā)送message.send.max.retries次后記錄該異常并繼續(xù)發(fā)送后續(xù)數(shù)據(jù)。目的：kafka為預(yù)防一旦某一個(gè)Broker宕機(jī)，則其上所有的Partition數(shù)據(jù)都不可被消費(fèi)；同時(shí)提高了整個(gè)系統(tǒng)的可用性。Replica均勻分布到整個(gè)集群Kafka分配Replica的算法如下：設(shè)有n個(gè)broker，i個(gè)partion，j個(gè)Replica，將broker和parttion排序后。將第i個(gè)Partition分配到第（imodn）個(gè)Broker上將第i個(gè)Partition的第j個(gè)Replica分配到第（(i+j)modn）個(gè)Broker上kafka的數(shù)據(jù)同步機(jī)制kafka的各個(gè)分區(qū)備份之間的同步機(jī)制：1.生產(chǎn)者發(fā)送消息到leader，leaderlog數(shù)據(jù)2.flowller從leader拉取消息。3.flower接收到消息，放到內(nèi)存ackleader。4.leaderackproducer。Kafka之Replicationtools最優(yōu)副本選舉工具bin/kafka-preferred-replica-election.s--zookeeper

localhost:12913/kafka

--path-to-json-file

topicPartitionList.jsonTopic和Partion轉(zhuǎn)移工具#轉(zhuǎn)移topic到新的brokerkafka-reassign-partitions.sh

--topics-to-move-json-file

topics-to-move.json

--broker-list

"5,6,7"

--execute

#轉(zhuǎn)移partition的一些replica到指定的brokerkafka-reassign-partitions.sh

--manual-assignment-json-file

partitions-to-move.json

--execute

kafka磁盤上限測試kafka磁盤容量測試，當(dāng)一個(gè)topic的一個(gè)分區(qū)所在的磁盤用盡，是否會(huì)異常？測試環(huán)境：1.開啟一個(gè)寫kafka的生生產(chǎn)者，向kafka寫消息，每條消息10kb。2.Topic為1個(gè)partitions，2個(gè)Replica；Replica對(duì)應(yīng)的磁盤大小為5