我們一起聊聊 B 站 Kafka 探索與實(shí)踐
1.背景
Kafka 是我們公司各個(gè)部門的重要數(shù)據(jù)中間件，主要用于上報(bào)、暫存和分發(fā)各種數(shù)據(jù)。我們不僅支持大數(shù)據(jù)場(chǎng)景，同時(shí)也保障在線場(chǎng)景的穩(wěn)定性。
我們擁有1000+臺(tái)Kafka機(jī)器，組成了20多個(gè)集群，針對(duì)不同業(yè)務(wù)場(chǎng)景配置了不同規(guī)格的機(jī)器，磁盤類型包括HDD，SSD和NVME。每日PB級(jí)輸入，輸出數(shù)十PB，肩負(fù)著公司數(shù)據(jù)傳輸?shù)闹厝巍ｋS著集群規(guī)模的擴(kuò)張，我們也遇到了越來(lái)越多的挑戰(zhàn)。
圖片
2.面臨的挑戰(zhàn)與痛點(diǎn)
- 客戶端的讀寫方式多種多樣，難以預(yù)測(cè)，集群的穩(wěn)定性和資源利用率難以協(xié)調(diào)。過(guò)高的讀寫操作可能會(huì)導(dǎo)致機(jī)器磁盤I/O爆滿，影響用戶的讀寫體驗(yàn)。
- 集群多業(yè)務(wù)共用，如何減少核心業(yè)務(wù)與普通業(yè)務(wù)的互相影響，減少爆炸半徑。
- 開(kāi)源版本雖然有限速的功能，但是限速的粒度很粗，不夠靈活，難以實(shí)時(shí)根據(jù)磁盤狀態(tài)進(jìn)行對(duì)應(yīng)調(diào)整。
- 為保證集群穩(wěn)定性，Kafka機(jī)器上下線流程繁瑣、效率低下，如何提升上下線效率。
- 開(kāi)源版本在分配partition時(shí)只考慮每臺(tái)機(jī)器Partition數(shù)量，不考慮磁盤流量負(fù)載情況，也不考慮Topic之間的流量差異，會(huì)造成集群中機(jī)器間、磁盤間負(fù)載不均衡。
- 開(kāi)源版本缺少自動(dòng)均衡和遷移速率控制的功能，這可能會(huì)對(duì)用戶的實(shí)時(shí)讀寫造成影響。人工進(jìn)行遷移和速率控制的操作不僅效率低，而且難以管理。
- 隨著公司業(yè)務(wù)不斷擴(kuò)大，一套IDC不足以支撐整個(gè)公司的服務(wù)，如何控制協(xié)調(diào)多個(gè)IDC是所面臨的問(wèn)題。
- Kafka 只有一個(gè)工作線程池，慢請(qǐng)求可能導(dǎo)致線程池阻塞，影響其他請(qǐng)求處理效率。
3.思考與方案
3.1 Guardian - Kafka federation
cluster controller
- 由于Kafka承載著公司傳輸?shù)闹匾氊?zé)，集群規(guī)模的不斷擴(kuò)大給集群的穩(wěn)定性和易治理性帶來(lái)了巨大的挑戰(zhàn)。為了應(yīng)對(duì)這些挑戰(zhàn)，我們需要開(kāi)發(fā)一套自動(dòng)化治理系統(tǒng)，來(lái)有效地解決目前面臨的問(wèn)題。
- Guardian 是一套自研的Kafka federation cluster controller。該服務(wù)通過(guò)Raft保證了高可用和一致性。此外該服務(wù)會(huì)從Kafka Server端收集各類數(shù)據(jù)進(jìn)行計(jì)算和分析，執(zhí)行治理計(jì)劃。包含以下功能
1. federation cluster的元數(shù)據(jù)管理與集群
2. remote storage元數(shù)據(jù)管理與存儲(chǔ)
3. uuid(topicId, segementId)的分配
4. 收集集群信息進(jìn)行調(diào)度
5. 多租戶管理與label隔離
6. 故障預(yù)警與自愈
圖片
- 備注：基于JMX協(xié)議采集Metrics的性能非常差，這是因?yàn)閖mx協(xié)議一個(gè)請(qǐng)求只能獲取一個(gè)mbean。隨著metric的加強(qiáng)，可能達(dá)到萬(wàn)級(jí)別，此時(shí)cpu消耗會(huì)占20%甚至更高。Kafka Reporter為基于GRPC，http協(xié)議的內(nèi)置Metric上報(bào)服務(wù)，只需要一個(gè)rpc即可拉取全部監(jiān)控?cái)?shù)據(jù)。
3.2 集群層面治理
3.2.1 Partition級(jí)別限速保護(hù)
相關(guān)背景
- Kafka 是一個(gè) I/O 密集型的服務(wù)，用戶行為不可預(yù)測(cè)切多變。當(dāng)用戶讀取最新數(shù)據(jù)時(shí)，能夠從 Page Cache 中高效快速獲取，但如果需要從磁盤中讀取數(shù)據(jù)，那么就要考慮如何控制磁盤 I/O 和磁盤資源的使用，以便為用戶提供最大的吞吐量。
- 開(kāi)源的限速方案存在問(wèn)題是粗力度，比如限制某ClientId的讀速度為5MB/s，那么此Client在某臺(tái)機(jī)器下讀取所有訂閱的Partition的速度的總和被限制到5MB/s左右，無(wú)法精確限制到Partition。
相關(guān)優(yōu)化
- 我們的目標(biāo)是盡可能讓磁盤合理且充分地利用。因?yàn)镻artition是落在一個(gè)個(gè)磁盤下，為了給用戶提供更高的吞吐，我們需要將每個(gè)磁盤的資源利用率進(jìn)一步提高，因此我們給Kafka新增Partition粒度的限速邏輯。為了保證磁盤的正常運(yùn)行，我們通過(guò)管控系統(tǒng)實(shí)時(shí)監(jiān)測(cè)每個(gè)磁盤的ioutil和Latency等關(guān)鍵指標(biāo)。一旦發(fā)現(xiàn)某塊磁盤超過(guò)了我們?cè)O(shè)定的閾值，我們就會(huì)判斷該磁盤的健康度下降，需要進(jìn)行優(yōu)化處理。因此，我們會(huì)嘗試對(duì)該磁盤下的io操作進(jìn)行限速，使得該磁盤的ioutil和Latency恢復(fù)到合理水平。是否讀取磁盤數(shù)據(jù)難以判斷，我們實(shí)現(xiàn)了一個(gè)估算算法，用可用內(nèi)存大小除以磁盤讀寫的速率和，粗略估算出一段數(shù)據(jù)能在PageCache里面存在的時(shí)間T，使用Partition的MessageInRate * T 估算出此Partition能在PageCache里面緩存數(shù)據(jù)條數(shù)，用此Partition LEO - MessageInRate * T 如果大于要拉取數(shù)據(jù)的Offset，那么認(rèn)為此數(shù)據(jù)是實(shí)際讀磁盤數(shù)據(jù)。
- Guardian基于集群監(jiān)控?cái)?shù)據(jù)進(jìn)行集群磁盤的健康度檢查，根據(jù)每一個(gè)分區(qū)的監(jiān)控?cái)?shù)據(jù)進(jìn)行根因分析，根據(jù)分析結(jié)果及時(shí)進(jìn)行限速調(diào)整
- 磁盤io相關(guān)行為分為六種：用戶讀/寫磁盤，主從同步讀/寫磁盤和磁盤間遷移讀/寫磁盤。
- 異常行為：超過(guò)預(yù)期的寫磁盤，任何讀磁盤。
- 對(duì)所有異常行為排序的隊(duì)列我們稱之為”異常行為隊(duì)列“。異常行為隊(duì)列排序方式為當(dāng)前流量大到小排優(yōu)先級(jí)。
- 具體流程入下圖所示：
圖片
3.2.1.3 Partition級(jí)別限速效果：
- 無(wú)限速保護(hù)，直接安排磁盤搬遷任務(wù)：
圖片
圖片
- 在開(kāi)啟Partition級(jí)別讀磁盤限速保護(hù)后安排磁盤間搬遷任務(wù)：
圖片
圖片
- 在實(shí)際使用中，根據(jù)集群狀況實(shí)時(shí)進(jìn)行自動(dòng)限速保護(hù)的功能對(duì)Kafka集群的穩(wěn)定性做出了很大貢獻(xiàn)。
3.2.2 自動(dòng)Partition均衡
相關(guān)背景
- 為了解決開(kāi)源 Kafka 的負(fù)載分配不均勻?qū)е碌拇疟P熱點(diǎn)問(wèn)題，我們開(kāi)發(fā)了一種基于磁盤指標(biāo)、集群 Topic 分布情況和每個(gè) Partition 流量指標(biāo)的 Partition 自動(dòng)均衡遷移計(jì)劃功能。
圖片
相關(guān)優(yōu)化（Kafka版本為2.4及以上）
- 遷移計(jì)劃。我們根據(jù)采集到的數(shù)據(jù)進(jìn)行計(jì)算，分析哪些機(jī)器負(fù)載過(guò)高，篩選出需要均衡的Partition。生成遷移計(jì)劃時(shí)，會(huì)考慮目標(biāo)機(jī)器磁盤的流量負(fù)載，通過(guò)指標(biāo)采集模塊獲取每個(gè)磁盤的歷史窗口流量負(fù)載，然后按磁盤按照歷史流量負(fù)載中位數(shù)進(jìn)行排序，選擇最小的磁盤放入分區(qū)副本。
- 搬遷任務(wù)增量提交，解決長(zhǎng)尾問(wèn)題。在集群執(zhí)行均衡遷移計(jì)劃時(shí)，不同Partition所承載的流量不同，所在的機(jī)器負(fù)載不同，搬遷所需要的時(shí)間也會(huì)不同。我們采取增量提交均衡任務(wù)，讓耗時(shí)較長(zhǎng)的任務(wù)不會(huì)阻塞其他任務(wù)，保持高效執(zhí)行。
- 自動(dòng)動(dòng)態(tài)調(diào)整搬遷速度。一天內(nèi)不同時(shí)間段集群的負(fù)載是不同的，我們會(huì)根據(jù)集群內(nèi)機(jī)器的負(fù)載動(dòng)態(tài)調(diào)整搬遷的速度，讓搬遷任務(wù)不會(huì)影響到集群的穩(wěn)定性和用戶的使用。
- 多并發(fā)搬遷。我們支持不同集群不同Partition并發(fā)的執(zhí)行搬遷計(jì)劃，并可控制并發(fā)度。
- Partition預(yù)分配。新建Topic時(shí)，我們會(huì)根據(jù)當(dāng)前磁盤負(fù)載和該Topic的預(yù)期估值流量進(jìn)行計(jì)算，生成該Topic 所有Partition的預(yù)分配計(jì)劃。
- Leader均衡。為了避免Partition Leader機(jī)器嚴(yán)重不均衡造成的機(jī)器熱點(diǎn)問(wèn)題，我們會(huì)自動(dòng)生成均衡計(jì)劃進(jìn)行Partition Leader均衡。
- 節(jié)點(diǎn)異常、Topic擴(kuò)容 遷移取消。當(dāng)某個(gè)節(jié)點(diǎn)異常，搬遷計(jì)劃無(wú)法順利執(zhí)行完成時(shí)或當(dāng)流量上漲需要進(jìn)行Partition擴(kuò)容時(shí)（開(kāi)源Kafka不支持在搬遷時(shí)進(jìn)行擴(kuò)容），我們會(huì)自動(dòng)取消當(dāng)前所影響的搬遷計(jì)劃。
3.2.3 多租戶資源隔離管理
相關(guān)背景
- 由于各業(yè)務(wù)場(chǎng)景及優(yōu)先級(jí)不同，我們需要盡可能避免互相影響及互相爭(zhēng)搶資源的狀況，同時(shí)還要避免集群數(shù)量膨脹帶來(lái)的運(yùn)維及使用問(wèn)題。由此提出Kafka多租戶方案，提高對(duì)共享資源的管控，對(duì)高優(yōu)流提供更好的隔離性。
圖片
相關(guān)優(yōu)化
- 資源獨(dú)占
1） Topic創(chuàng)建
a. 相同業(yè)務(wù)域的Topic共享獨(dú)占資源，用戶在創(chuàng)建Topic的時(shí)候指定對(duì)應(yīng)的獨(dú)占資源即可，該Topic僅會(huì)在獨(dú)占資源上創(chuàng)建。支持動(dòng)態(tài)調(diào)整，任務(wù)無(wú)感。
2）機(jī)器劃分
a. 新的獨(dú)占資源: 從資源池劃所需機(jī)器。
b. 已存在獨(dú)占資源 : 檢查已有資源是否滿足新需求，不滿足時(shí)按需擴(kuò)容。
c. 獨(dú)占資源縮容 : 選擇機(jī)器，將對(duì)應(yīng)機(jī)器上的所有partition遷移到剩余的機(jī)器后，機(jī)器歸還資源池。
3）支持自動(dòng)限速保護(hù)、自動(dòng)均衡搬遷功能。
- 這里的租戶概念對(duì)應(yīng)為應(yīng)用域劃分。
3.2.4 多機(jī)房管理
相關(guān)背景
- 隨著公司的業(yè)務(wù)拓展，單個(gè)IDC不足以支持需求，因此系統(tǒng)需要支持跨idc管理。
圖片
相關(guān)優(yōu)化
- 業(yè)務(wù)遷移idc后，秉承著就近原則，對(duì)應(yīng)的數(shù)據(jù)也需要遷移idc。我們的系統(tǒng)支持使用方無(wú)感的情況下，一鍵完成不同IDC間的Topic遷移。
按需配置副本放置策略，系統(tǒng)根據(jù)所填配置自動(dòng)生成遷移計(jì)劃，在遷移的過(guò)程中，有限速模塊保證集群的穩(wěn)定性，遷移過(guò)程中無(wú)需人工干預(yù)。
配置粒度可控，從集群級(jí)別到topic級(jí)別均可。
Topic在遷移過(guò)程中，offset保持一致，期間用戶讀寫均無(wú)感。
- 一般相關(guān)的業(yè)務(wù)在同一IDC中，但是也會(huì)有多個(gè)任務(wù)分布在不同機(jī)房。我們支持idc感知的從讀功能，用戶請(qǐng)求時(shí)返回就近的副本，在同IDC讀取。
圖片
- 通過(guò)上報(bào)數(shù)據(jù)分析，實(shí)時(shí)計(jì)算專線消耗，保護(hù)專線。
3.2.5 請(qǐng)求隊(duì)列拆分
相關(guān)背景
一般物理機(jī)上有多塊磁盤，有較大的內(nèi)存和較大的CPU。不同公司Kafka的部署方式不太一樣，有的選擇部署多個(gè)實(shí)例，一個(gè)實(shí)例一塊磁盤，有的則是部署一個(gè)實(shí)例，利用Kafka支持多盤的特性。兩種方案各有利弊，我們采用的是后者。一臺(tái)物理機(jī)上有10+HDD盤。一個(gè)實(shí)例上有上千Partition，上萬(wàn)個(gè)連接。如果其中一塊磁盤Fail-Slow，或負(fù)載特別高，磁盤上的請(qǐng)求耗時(shí)會(huì)增長(zhǎng)很多。假設(shè)請(qǐng)求均勻，受影響比例理應(yīng)僅為1/14 = 7%，但實(shí)際會(huì)有100%影響（討論的Kafka版本為2.4.1）。
圖片
- 上圖為Kafka相關(guān)部分的線程模型，涉及的文件有：
- core/src/main/scala/kafka/network/SocketServer.scala
- clients/src/main/java/org/apache/kafka/common/network/Selector.java
- clients/src/main/java/org/apache/kafka/common/network/KafkaChannel.java
- core/src/main/scala/kafka/server/KafkaRequestHandler.scala
Kafka雖沒(méi)使用Netty框架，架構(gòu)是大同小異的，依然是N個(gè)select/poll線程（下文稱為Processor線程）處理network io + m個(gè)工作線程（下文稱為requestHandler線程）。如上文所述，多數(shù)情況可以很好工作。
Kafka作為一種分布式消息系統(tǒng)的實(shí)現(xiàn)，最主要的工作是收發(fā)消息(讀寫文件)，以下我們分別討論P(yáng)roduce和Consume的部分細(xì)節(jié)。
Produce：
- Acceptor線程： 建立連接后，該連接被投遞到了某一個(gè)Processor中（roundRobin）。
- Processor線程：Client發(fā)送請(qǐng)求后，Read Socket，讀到一個(gè)完整的請(qǐng)求包后將其后放入到requestChannel中。
- 某一個(gè)requestHandler線程：Poll到該Req，經(jīng)過(guò)一系列的校驗(yàn)，定位到Partition對(duì)應(yīng)的File和當(dāng)前Position后寫入，滿足一定條件時(shí)調(diào)用Flush。假設(shè)ACK = 1，既Leader寫入后就返回成功，生成Response對(duì)象（包含已序列化的byte[]）寫入到對(duì)應(yīng)Processor的responseQueue中。
- Processor線程：獲取到Response后，Write socket （省略select/poll及粘拆包細(xì)節(jié)）。
機(jī)械磁盤的寫是非常慢的（混合隨機(jī)讀寫場(chǎng)景），實(shí)際上多數(shù)produce請(qǐng)求延遲很低，這是因?yàn)閘inux的page cache機(jī)制將寫磁盤操作轉(zhuǎn)化為寫內(nèi)存，而寫內(nèi)存是非常快的。只有在極個(gè)別的情況下（主動(dòng)fsync，或者達(dá)到閾值），才會(huì)同步寫磁盤。這也是Kafka的produce latency存在毛刺（很多時(shí)候看上去有規(guī)律），且毛刺遠(yuǎn)大于均值的原因。
Consume：
- Acceptor線程： 建立連接后，該連接被投遞到了某一個(gè)Processor中（RoundRobin）。
- Processor線程：Client發(fā)送請(qǐng)求后，Read Socket，讀到一個(gè)完整的請(qǐng)求包后將其后放入到requestChannel中。
- 某一個(gè)requestHandler線程：Poll到該Req，經(jīng)過(guò)一系列的校驗(yàn)，根據(jù)FetchOffset找到對(duì)應(yīng)的Log File Position（先通過(guò)Index找到大致位置，然后一個(gè)個(gè)Batch的讀取，直到找到FetchOffset所在的Batch的 Position），假設(shè)數(shù)據(jù)足夠多，生成Response（保存了File Position Size的元信息）寫入到對(duì)應(yīng)Processor的ResponseQueue中。
- Processor線程：獲取到response后，調(diào)用Sendfile(Zero Copy)發(fā)送 （省略select/poll及粘拆包細(xì)節(jié)）。
注意：在requestHandler和Processor中都有io操作。
類似于Produce，得益于Page Cache機(jī)制，大部分讀操作僅需要讀內(nèi)存，大多數(shù)場(chǎng)景下耗時(shí)非常短。
Batch粒度是用戶寫時(shí)決定的，如果用戶Batch寫入條數(shù)過(guò)小，就會(huì)大大增加Offset尋址次數(shù)。在Page Cache Missing時(shí)還會(huì)產(chǎn)生不可忽略的io負(fù)載。（參考core/src/main/scala/Kafka/log/LogSegment.scala LogSegment.translateOffset）。
雖然Zero Copy有上下文切換以及內(nèi)存Copy的優(yōu)勢(shì)，可仍舊是同步io。Page Cache Missing時(shí)需要從磁盤讀取到Page Cache，而一個(gè)Processor負(fù)責(zé)處理若干Socket的收發(fā)，Processor線程的阻塞意味著這些Socket上的Requests也需要忍受額外的Latency。
另一個(gè)需要注意的點(diǎn)是使用低版本的Client與高版本的Server交互時(shí)，Server需要做Up/Down（Produce/Consume） Conversion來(lái)適配Client Version。此時(shí)無(wú)法直接調(diào)用Sendfile發(fā)送，而是一個(gè)個(gè)Batch的讀取，轉(zhuǎn)換，然后寫入Socket（參考clients/src/main/java/org/apache/kafka/common/record/LazyDownConversionRecords.java AbstractIterator.makeNext）。
相關(guān)優(yōu)化
- 優(yōu)化基于以下：
每個(gè)Client實(shí)例（連接）生產(chǎn)特定少數(shù)Topic，消費(fèi)特定少數(shù)Partition。
如果能做到線程隔離，保證所有的慢請(qǐng)求與正常請(qǐng)求不共享線程，則正常請(qǐng)求不會(huì)收到影響。
慢請(qǐng)求總是集中在部分Partition上，且總是少數(shù)。
- 線程模型修改如上圖
ChannelMarker，監(jiān)控每個(gè)Channel（連接）的RequestHandler線程耗時(shí)及Processor線程耗時(shí)并打分，標(biāo)記較慢的channel。
RequestHandler根據(jù)requestType拆分線程池，F(xiàn)etch，Produce，Default。并且新增對(duì)應(yīng)的SlowRequest的線程池。1.中標(biāo)記的channel請(qǐng)求會(huì)投遞到對(duì)應(yīng)的slowRequest線程池中，保證線程隔離。
如果ChannelMarker不能很好分類，該算法近似于原本的線程模型。另外將其他請(qǐng)求與讀寫線程池拆分，能盡可能的避免Controller與Admin請(qǐng)求超時(shí)。
3.2.6 Tiered-Storage
相關(guān)背景
- Kafka的Partition和磁盤上的目錄一一對(duì)應(yīng)，因此有以下問(wèn)題:
個(gè)別磁盤讀寫量過(guò)大，會(huì)出現(xiàn)磁盤熱點(diǎn)，受限于磁盤性能上限。
遷移需要同步完整數(shù)據(jù)，導(dǎo)致遷移時(shí)間非常長(zhǎng)，需要小時(shí)級(jí)甚至天級(jí)，提前治理要求極高。
Partition不支持讀寫資源分離，讀寫磁盤會(huì)互相影響。
Partition的總數(shù)據(jù)量取決于磁盤的容量上限，無(wú)法滿足1周及以上的需求。
任務(wù)間的寫入消費(fèi)速度不同，數(shù)據(jù)過(guò)期時(shí)間不同和峰谷流量情況不同，導(dǎo)致難以合理規(guī)劃。
- 社區(qū)提供了一套Kafka Tired Storage 方案，但是也存在很多不足：
社區(qū)主推S3作為Remote Storage，缺少基于HDFS的實(shí)現(xiàn)。
目前kafka提供的實(shí)現(xiàn)是TopicBased，1 metadata per message，且為了保證元數(shù)據(jù)不丟需要等message發(fā)送完成后才能繼續(xù)工作，partition的增加會(huì)加劇該實(shí)現(xiàn)的性能問(wèn)題。
當(dāng)發(fā)生leader stale時(shí)，stale leader依然能夠?qū)憈opic，導(dǎo)致臟數(shù)據(jù)。并且基于compact topic的數(shù)據(jù)清理也有性能和及時(shí)性兩方面的問(wèn)題
remoteLogManager會(huì)實(shí)時(shí)消費(fèi)metadata topic，忽略本replica無(wú)關(guān)的partition。reassign partition不會(huì)從頭消費(fèi)導(dǎo)致部分metadata丟失。
治理要求依然很高，同步副本數(shù)據(jù)耗時(shí)依然需要30min+。
注：截止今天，kafka已發(fā)布（3.6.0）的remote storage相關(guān)功能只是early access version，后續(xù)可能優(yōu)化。
圖片
相關(guān)優(yōu)化
- 我們的目標(biāo)是將Kafka的Partition和磁盤強(qiáng)綁定關(guān)系打破，允許同一個(gè)Partition的數(shù)據(jù)分布在不同Storage層上，并且我們采用HDFS作為Remote Storage。
圖片
- 我們?cè)O(shè)計(jì)了分層存儲(chǔ)meta服務(wù)（基于Raft實(shí)現(xiàn)HA）
- 官方設(shè)計(jì)的是通過(guò)meta topic來(lái)存儲(chǔ)meta，但是該方案有個(gè)大問(wèn)題，無(wú)法保證數(shù)據(jù)不丟，且無(wú)法高效獲取某個(gè)partition的meta（需要回放全部數(shù)據(jù)）。我們傾向于實(shí)現(xiàn)自己的Meta Server。Meta Server內(nèi)部是通過(guò)Raft做到一致性及HA的。與Kafka交互的協(xié)議上選擇GRPC。Meta Server在功能上類似于meta topic，每一組復(fù)制，刪除都是一個(gè)事件組（metaData）, 每一個(gè)事件組都有自增且唯一的編號(hào)，拉取Event 通過(guò)這個(gè)編號(hào)獲取。Meta server底層用RocksDB存儲(chǔ)，Key為cluster_topic_uuid_id_partition_event_id，Value為事件組的PB序列化Byte，拉取通過(guò)rangeScan實(shí)現(xiàn)即可。snapshot保存尚未刪除的復(fù)制信息。
- 支持基于不同策略的offset拉取模式。
- 當(dāng)新的Follower拉取數(shù)據(jù)時(shí)，返回local-log-start-offset。我們自定義了local-start-fetch-offset。該值為 max （ local-log-start-offset，local-log-end-offset - 平均速度 * 閾值 。為什么不用remote-log-end-offset，有可能該值和local-log-end-offset 過(guò)于接近，導(dǎo)致過(guò)多請(qǐng)求從Remote Storage讀取，增加了RS的開(kāi)銷。為什么不用local-log-start-offset，一般我們本地仍會(huì)保存若干小時(shí)數(shù)據(jù)，全部數(shù)據(jù)都備份仍舊需要相當(dāng)?shù)臅r(shí)間，且這些數(shù)據(jù)都是在RS上的。出于提速的考量只需要同步local-start-fetch-offset的數(shù)據(jù)。
- 大于local-start-fetch-offset的數(shù)據(jù)，我們通過(guò)Local Read獲得，在此之前的數(shù)據(jù)通過(guò)Remote Storage獲得。Leader尚未完成Meta信息同步時(shí)，則Local有的數(shù)據(jù)都從Local read。Local 沒(méi)有的數(shù)據(jù)（小于local-start）由RLM代理嘗試獲取，直到超時(shí)。同步完成但是Remote也沒(méi)有時(shí)OffsetOutOfRange ，Remote有但是Remote故障時(shí)等待直至超時(shí)。
- 修改協(xié)議支持批量，不再是寫一個(gè)Segment寫一次Meta，增加了Leader Fence保證強(qiáng)一致。
- Segment下載到本地Cache后讀取，HDFS 1 segment/per read, 大幅降低HDFS負(fù)擔(dān)和時(shí)延要求。
- 減少寫RT抖動(dòng)，降低寫延遲 （因?yàn)镵afka Broker不再需要存儲(chǔ)全部數(shù)據(jù)了，SSD磁盤的容量限制不再是瓶頸，而SSD的低延時(shí)卻有極大的優(yōu)勢(shì)）
3.2.7 Kafka 審計(jì)功能
相關(guān)背景
- 公司內(nèi)部使用Kafka的部門非常多，一個(gè)Topic經(jīng)常多個(gè)用戶在使用，無(wú)法精準(zhǔn)定位到該Topic的上下游使用者，日常運(yùn)維難以治理和管控。
圖片
相關(guān)優(yōu)化
- 我們對(duì) Kafka 進(jìn)行了改造，增加了審計(jì)功能，可以提供生產(chǎn)、消費(fèi)等一系列請(qǐng)求的詳細(xì)數(shù)據(jù)，并將這些數(shù)據(jù)實(shí)時(shí)寫入到 Clickhouse 中，方便問(wèn)題排查。
- 除此之外，我們支持查詢某個(gè)時(shí)間切片內(nèi)的機(jī)器和集群的詳細(xì)情況。
- 基于審計(jì)數(shù)據(jù)，我們實(shí)現(xiàn)了成本管理系統(tǒng)，可以有效地治理一些冗余的 Topic，達(dá)到降本增效的目的。
- 利用 Kafka 的審計(jì)功能，輔助修正了歷史上混亂的 Topic 權(quán)限，提升了數(shù)據(jù)的安全性。
3.3 運(yùn)維層面治理
3.3.1 集群平滑發(fā)布
相關(guān)背景
- 隨著集群規(guī)模不斷地?cái)U(kuò)大，如何快速平滑的發(fā)布成為一個(gè)越來(lái)越大的難題。
圖片
相關(guān)優(yōu)化
- 我們開(kāi)發(fā)了一種能夠自動(dòng)批量上下線機(jī)器的服務(wù)，用戶只需選擇目標(biāo)集群，服務(wù)就會(huì)根據(jù)集群的實(shí)時(shí)狀態(tài)，在不影響集群服務(wù)穩(wěn)定性的情況下，自動(dòng)完成機(jī)器的上下線操作。整個(gè)過(guò)程無(wú)需人工干預(yù)。
- 機(jī)器下線
- 當(dāng)某臺(tái)機(jī)器需要下線時(shí)，服務(wù)會(huì)先將該機(jī)器作為L(zhǎng)eader的所有Partition進(jìn)行遷移，確保Partition的可用性不受影響。
- 機(jī)器上線
- 當(dāng)某臺(tái)機(jī)器恢復(fù)正常并重新提供服務(wù)時(shí)，服務(wù)會(huì)自動(dòng)將該機(jī)器原先負(fù)責(zé)的所有Partition恢復(fù)到原來(lái)的狀態(tài)。
結(jié)果：極大節(jié)約人力成本，Kafka滾動(dòng)升級(jí)從15人天縮短到1人小時(shí)(提單后自動(dòng)化)、搬遷完全自動(dòng)化，降低對(duì)業(yè)務(wù)的影響。
4.1 未來(lái)展望
- 支持分鐘級(jí)調(diào)度。目前遷移的耗時(shí)一般在小時(shí)級(jí)別到天級(jí)，下一步是將遷移任務(wù)進(jìn)一步提速，使任務(wù)在分鐘級(jí)完成。
- 支持分鐘級(jí)自檢自愈系統(tǒng)。目前硬件故障需要人為操作移除故障節(jié)點(diǎn)，下一步目標(biāo)是全自動(dòng)分鐘級(jí)摘除。
- 支持動(dòng)態(tài)擴(kuò)縮容。目前集群擴(kuò)縮容完全依賴人為操作，未來(lái)集群的擴(kuò)縮容將完全自動(dòng)化，服務(wù)根據(jù)集群實(shí)時(shí)狀態(tài)進(jìn)行分析后自動(dòng)進(jìn)行機(jī)器擴(kuò)縮容。
本期作者
張辰安嗶哩嗶哩資深開(kāi)發(fā)工程師
劉博為 嗶哩嗶哩高級(jí)開(kāi)發(fā)工程師
