之前為大家分享了不少 Kafka 原理解析類的干貨，今天咱們一起來看看 360 基于 Kafka 千億級(jí)數(shù)據(jù)量的深度實(shí)踐!

[[286310]]

圖片來自 Pexels 

本文主要圍繞如下內(nèi)容分享：

• 消息隊(duì)列選型
  
• Kafka 在 360 商業(yè)化的現(xiàn)狀
  
• Kafka Client 框架
  
• 數(shù)據(jù)高可用
  
• 負(fù)載均衡
  
• 鑒權(quán)、授權(quán)與 ACL 方案
  
• Quota 機(jī)制
  
• 跨 IDC 的數(shù)據(jù)同步
  
• 監(jiān)控告警
  
• 線上問題及解決方案

消息隊(duì)列選型

當(dāng)時(shí)主要考慮以下幾個(gè)維度：

• 社區(qū)活躍度
• 客戶端支持
• 吞吐量

對(duì)比幾個(gè)系統(tǒng)下來，覺得 Kafka 比較符合我們的要求?，F(xiàn)在有一個(gè)新的開源系統(tǒng) Pulsar，我覺得也可以嘗試一下。

 

Kafka 設(shè)計(jì)上的亮點(diǎn)如下：

 

Kafka 性能和吞吐都很高，通過 Sendfile 和 Pagecache 來實(shí)現(xiàn) Zero Copy 機(jī)制，順序讀寫的特性使得用普通磁盤就可以做到很大的吞吐，相對(duì)來說性價(jià)比比較高。

Kafka 通過 Replica 和 ISR 機(jī)制來保證數(shù)據(jù)的高可用。

Kafka 集群有兩個(gè)管理角色：

• Controller 主要是做集群的管理。
• Coordinator 主要做業(yè)務(wù)級(jí)別的管理。

這兩種角色都由 Kafka 里面的某個(gè) Broker 來?yè)?dān)任，這樣 Failover 就很簡(jiǎn)單，只需要選一個(gè) Broker 來替代即可。

從這個(gè)角度來說 Kafka 有一個(gè)去中心化的設(shè)計(jì)思想在里面， 但 Controller 本身也是一個(gè)瓶頸，可以類比于 Hadoop 的 Namenode。

CAP 理論相信大家都有了解過，分布式系統(tǒng)實(shí)現(xiàn)要么是 CP，要么是 AP。

Kafka 實(shí)現(xiàn)比較靈活，不同業(yè)務(wù)可以根據(jù)自身業(yè)務(wù)特點(diǎn)來對(duì) Topic 級(jí)別做偏 CP 或偏 AP 的配置。

支持業(yè)務(wù)間獨(dú)立重復(fù)消費(fèi)，并且可以做回放。

 

這個(gè)是 Kafka 的簡(jiǎn)要架構(gòu)，主要分為：

• 生產(chǎn)端
• Broker 端
• 消費(fèi)端

日志有三個(gè)層次：

• 第一個(gè)層次 Topic
• 第二個(gè)層次 Partition(每個(gè) Partition 是一個(gè)并行度)
• 第三個(gè)層次 Replica(Replica 表示 Partition 的副本數(shù))

Kafka 在 360 商業(yè)化的現(xiàn)狀

 

目前集群有千億級(jí)數(shù)據(jù)量，100 多臺(tái)萬(wàn)兆機(jī)器，單 Topic 的最大峰值 60 萬(wàn) QPS，集群的峰值大概在 500 萬(wàn) QPS。

 

我們的物理機(jī)配置 24Core/10G 網(wǎng)卡/128G 內(nèi)存/4T*12 HDD，值得說一下的是我們采用了萬(wàn)兆網(wǎng)卡加普通磁盤 4T*12 的配置，測(cè)下來磁盤吞吐和網(wǎng)絡(luò)吞吐是能夠匹配上的。

再者考慮到我們的數(shù)據(jù)量比較大，SSD 盤沒有特別大的且成本比較高。

磁盤的組織結(jié)構(gòu)我們用的是 JBOD，RAID10 也是很好的方案(磁盤成本會(huì)翻倍)。

我們目前的 Kafka 版本是 1.1.1，推薦大家部署 0.11 以上的版本會(huì)好一些，這個(gè)版本對(duì)協(xié)議做了很多優(yōu)化，對(duì)于后續(xù)的 2.x 版本都是兼容的。

這個(gè)是我們 Kafka 上下游相關(guān)的組件，生產(chǎn)端主要是各種 Kafka Clients/實(shí)時(shí)服務(wù)/Flume/Logstash。 

消費(fèi)端分為實(shí)時(shí)，離線(ETL)，監(jiān)控三部分。實(shí)時(shí)有 Spark/Flink/Storm 等主流框架, 離線部分我們基于 Flink 自研了一個(gè)統(tǒng)一落地框架 Hamal，從 Kafka 消費(fèi)一遍數(shù)據(jù)就可以落地到多個(gè)下游系統(tǒng)(HDFS、Hbase、Redis等)，可以避免重復(fù)消費(fèi)。

還有部分是監(jiān)控的需求，我們把 ES/InfluxDB 相關(guān)的日志打到 Kafka，然后再消費(fèi)出來通過 Grafana 展示，但目前我們已經(jīng)切到 Prometheus 上了。

Kafka Client 框架

為什么要做這個(gè)框架呢?之前有很多的業(yè)務(wù)部門用裸 API 自己去實(shí)現(xiàn) Kafka Client 的邏輯。

但是會(huì)有很多問題，有一些異常情況會(huì) Catch 不全，我們做這個(gè)框架是想把所有的細(xì)節(jié)屏蔽掉，然后暴露出足夠簡(jiǎn)單的接口。

這樣可以減少業(yè)務(wù)犯錯(cuò)的可能性，我們要確保極端的情況下比如網(wǎng)絡(luò)或集群異常時(shí)的可用性，如果網(wǎng)絡(luò)或集群不可用，數(shù)據(jù)會(huì)先落到本地，等恢復(fù)的時(shí)候再?gòu)谋镜卮疟P恢復(fù)到 Kafka 中。

 

我們實(shí)現(xiàn)了兩個(gè)框架：

• LogProducer，支持 at least once。
• LogConsumer，支持 at least once 和 exactly once 兩種語(yǔ)意，其中 exactly once 需要業(yè)務(wù)去實(shí)現(xiàn) Rollback 接口。

 

LogProducer 框架的大體思路是通過內(nèi)存隊(duì)列將日志發(fā)送到 Kafka，當(dāng) Kafka 或網(wǎng)絡(luò)不可用的情況下會(huì)寫本地磁盤，同時(shí)會(huì)有一個(gè)線程去實(shí)時(shí)檢測(cè) Kafka 或者網(wǎng)絡(luò)的可用情況，如果恢復(fù)就會(huì)加載磁盤日志并發(fā)送到 Kafka。

我們還支持一種共享內(nèi)存的策略來代替內(nèi)存，使用共享內(nèi)存是為了減少重啟過程中日志的丟失數(shù)。

 

LogConsumer 的框架實(shí)現(xiàn)，通過 Blocking Queue 將 Consumer 線程和 Worker 線程解耦，因?yàn)楝F(xiàn)實(shí)情況是消費(fèi)邏輯很簡(jiǎn)單，但是處理邏輯會(huì)很復(fù)雜。

這樣就可以對(duì) Consumer 線程和 Worker 線程做不同的配置，同時(shí)通過 Blocking Queue 還可以實(shí)現(xiàn)反壓機(jī)制。

比如 Worker 處理不過來了，這時(shí)候 Blocking Queue 就會(huì)滿，反壓到 Consumer 線程會(huì)停止消費(fèi)。

同時(shí)我們?cè)?Worker 線程接口里面會(huì)提供接口讓用戶提交到 global offsetmap。

如上圖我們提供三個(gè)組合接口，如果在業(yè)務(wù)處理與 Commit 中實(shí)現(xiàn)了業(yè)務(wù)端 Rollback 邏輯， 那么就是 exactly once 語(yǔ)義，默認(rèn)是 at least once 語(yǔ)義。

數(shù)據(jù)高可用

之前講過 Kafka 本身提供 Replica+ISR 的機(jī)制來保證數(shù)據(jù)高可用，但我們覺得這個(gè)可能還不夠，所以我們還要支持 Rack Aware。

比如 Replica=3 的情況，確保三個(gè)副本在不同的物理 Rack 上，這樣我們最多能容忍兩個(gè)物理機(jī)架同時(shí)出問題而數(shù)據(jù)仍可用，我們 Rack Aware 方案是與負(fù)載均衡方案一起做掉的，具體后面會(huì)講。

 

值得注意的是 Kafka 官方也支持 Rack Aware，通過在 Broker 端配置 broker.rack 參數(shù)可實(shí)現(xiàn)。

但有一個(gè)限制，必須為每個(gè) Rack 分配數(shù)量相同的 Brokers，否則會(huì)導(dǎo)致 Replica 分配傾斜，實(shí)際情況是 IDC 的 Rack 是很多的，分配到的物理機(jī)分布也可能很隨機(jī)。

一個(gè)可以參考的解決思路是采用虛擬 Rack Group 的概念，比如維護(hù) 3 個(gè)虛擬 Rack Group，申請(qǐng)到的物理機(jī)加入到這 3 個(gè) Group 中，并確保 Rack Group 間分配的物理機(jī)數(shù)量一致。

當(dāng)然 Rack Group 間物理機(jī)不應(yīng)存在有相同物理 Rack 的情況。

負(fù)載均衡

Kafka 的負(fù)載均衡功能在 Confluent 商業(yè)版本才支持，負(fù)載均衡本質(zhì)上來說是 Replica 分配均勻問題。

我們一開始想通過經(jīng)典一致性 Hash 來解決，如下圖：

 

然后我們發(fā)現(xiàn)經(jīng)典一次性 Hash 不能滿足我們的需求，比如要加一個(gè)節(jié)點(diǎn) node5，只能分擔(dān)節(jié)點(diǎn) node2 的部分負(fù)載，不能做全局節(jié)點(diǎn)的負(fù)載均衡。

 

于是我們基于虛擬節(jié)點(diǎn)的一次性 Hash 的算法實(shí)現(xiàn)了一個(gè)方案，如圖所示：相同的顏色對(duì)應(yīng)同一個(gè)物理機(jī)，Hash 環(huán)上的都是虛擬節(jié)點(diǎn)。

這里有四個(gè)物理節(jié)點(diǎn)，其中 node4 是我們新加的節(jié)點(diǎn)。通過虛擬節(jié)點(diǎn)可以把物理節(jié)點(diǎn)的負(fù)載足夠均衡地分散出去，所以當(dāng)我把 node4 加到 Hash 環(huán)上的時(shí)候，分擔(dān)了所有物理機(jī)的負(fù)載。

算法實(shí)現(xiàn)的步驟分為兩個(gè)大的步驟：

①新建 hash circle：通過 vnode_str(比如 hostname-v0)做一個(gè) MD5 的 Hash，得到虛擬節(jié)點(diǎn)的 vnode_key，再用 ring 字典來保存虛擬節(jié)點(diǎn)到物理節(jié)點(diǎn)的映射，同時(shí)將 vnode_key 加入到 sorted_keys 的 list 中。

②在 Hash 環(huán)中分配 Replica：將(topic_name+partition_num+replica_num)作為 Key 用相同的 MD5 Hash 算法得到 replica_key。

接著二分查找該 replica_key 在 sorted_keys 中的 Position, 最后用 Ring 字典來映射到物理機(jī) Node，至此 Replica 分配完成。

 

我們基于這個(gè)算法解決三個(gè)問題：

• 添加物理節(jié)點(diǎn)只需遷移很小一部分?jǐn)?shù)據(jù)。
• 對(duì)不同配置的物理機(jī)做權(quán)重設(shè)置，可以支持異構(gòu)集群的部署。
• 實(shí)現(xiàn) Replica 的 Rack Aware，物理節(jié)點(diǎn)上面會(huì)有 Rack 信息，在為 Replica 分配物理節(jié)點(diǎn)的時(shí)候會(huì)記錄已經(jīng)分配的 Rack 信息。

如果有重復(fù)的情況，就會(huì)把 vnode_key 找到 Position 的位置 +1 找下一個(gè)物理節(jié)點(diǎn)，我們會(huì)確保三個(gè) Replica 的物理 Rack 一定是不一樣的(假如 Replica=3)。

Leader Balance：這是一種快速且成本低的負(fù)載 Balance 方法，因?yàn)?Kafka 只有 Leader 提供讀寫，所以通過 Leader 切換是可以達(dá)到負(fù)載切換的效果的，由于只是 Leader 切換不涉及數(shù)據(jù)同步，因此這個(gè)代價(jià)是比較小的。

Disk Rebalance：這個(gè) Feature 需要 Kafka1.1.0 版本之后才支持，Kafka 提供了一些腳本和 API 可以做 Balance 操作, 其本質(zhì)也是生成 Replica Plan 然后做 Reassign。

鑒權(quán)、授權(quán)和 ACL 方案

如果是新集群比較推薦基于 SASL 的 SCRAM 方案，實(shí)施起來比較簡(jiǎn)單。

如果老集群想中途施行鑒權(quán)授權(quán)機(jī)制會(huì)比較困難，需要推各個(gè)業(yè)務(wù)去修改配置，同時(shí)切換的過程也很容易出問題。

下面介紹下我們實(shí)現(xiàn)的一個(gè)白名單機(jī)制來解決老集群的問題，首先將老業(yè)務(wù)加入到白名單中，讓新業(yè)務(wù)通過工單流程來申請(qǐng) Topics 和 Consumers 兩種資源權(quán)限并加到白名單里，定期監(jiān)測(cè)非法(沒有走工單)Topics，Consumers 資源。

同時(shí)將這些資源都 Deny 掉，這樣就收緊了 Topics 和 Consumer 讀寫權(quán)限的口子，同時(shí)原有業(yè)務(wù)不會(huì)有任何影響。

 

Quota 機(jī)制

 

Quota 主要是為了解決多個(gè)業(yè)務(wù)間資源搶占問題。Quota 類型有兩種：

• 一種是限制網(wǎng)絡(luò)帶寬。
• 一種是限制請(qǐng)求速率(限制 CPU)。

我們對(duì)業(yè)務(wù)做了三個(gè)優(yōu)先級(jí)設(shè)置：高，中，低優(yōu)先級(jí)，高優(yōu)先級(jí)不做限制，中優(yōu)先級(jí)可容忍 lag，低優(yōu)先級(jí)極端情況可停掉，通過工具可以批量限制某個(gè)優(yōu)先級(jí)的所有業(yè)務(wù)，可以確保高優(yōu)先級(jí)業(yè)務(wù)及集群的安全。

跨 IDC 的數(shù)據(jù)同步

 

首先我們?yōu)槭裁匆隹?IDC 的數(shù)據(jù)同步?沒做這個(gè)同步之前業(yè)務(wù)可能對(duì)數(shù)據(jù)的讀寫沒有一個(gè) IDC 的概念，所以很容易就會(huì)有跨 IDC 的讀寫，多個(gè)業(yè)務(wù)還可能有重復(fù) Consume 和 Produce。

這就造成跨 IDC 網(wǎng)絡(luò)的極大浪費(fèi)， 加上跨 IDC 的網(wǎng)絡(luò)并不穩(wěn)定，有時(shí)候會(huì)有一些異常，業(yè)務(wù)也不一定能很好處理。

 

為了解決以上問題，我們統(tǒng)一做了跨 IDC 的數(shù)據(jù)同步服務(wù)，首先我們約定業(yè)務(wù)只能做本 IDC 的讀寫，不允許做跨 IDC 的讀寫，如果有跨 IDC 的數(shù)據(jù)需求，要向我們申請(qǐng)，通過 Mirrormaker 去同步一份過來。

這樣做有兩個(gè)好處：

• 一是屏蔽了異常對(duì)業(yè)務(wù)的影響。
• 二是節(jié)省了 IDC 之間的帶寬(我們通過同步機(jī)制能保證這份數(shù)據(jù)只傳輸一份)。

我們還基于 Marathon/Mesos 對(duì)這個(gè)服務(wù)做了 Pass 化，提高了服務(wù)的 SLA。

 

監(jiān)控告警

 

我們的監(jiān)控警告平臺(tái)如下：

• 基于 Jmx exporter+Promehteus+Grafana 來做圖表展示，在每個(gè) Broker 上面部署 Jmx exporter，Prometheus 會(huì)去 Pull 這些數(shù)據(jù)，最后通過 Grafana 來展示。
• 基于 Kafka Manager 做瞬態(tài)指標(biāo)的監(jiān)控。
• 基于 Burrow 做 Consumer lag 的監(jiān)控。
• 基于 Wonder 來做告警，這個(gè)是 360 內(nèi)部實(shí)現(xiàn)的一個(gè)組件，類似 Zabbix。

 

線上問題及解決方案

 

磁盤故障：我們通過 Smartctl 來監(jiān)測(cè)，首先狀態(tài)是要 Passed 的，其次我們會(huì)判斷 197 Current_Pending_Sector 這個(gè)屬性值不能大于 100， 如果大于 100 這個(gè)磁盤可能有讀寫性能問題。

bootstrap.servers 性能瓶頸：該參數(shù)可以配置多臺(tái) Broker，這些 Broker 作為 Proxy 的角色為 Kafka Clients 提供 Lookup 服務(wù)。

如果集群規(guī)模很大，Clients 很多的情況下，這些 Proxy 角色的 Broker 的負(fù)載會(huì)很大，為了解決這個(gè)問題，我們對(duì) bootstrap.servers 參數(shù)做了 VIP 配置。

每個(gè) VIP 可以綁定任意多的 Brokers，這樣在客戶端不需要修改配置的情況下可以對(duì) Proxy 動(dòng)態(tài)擴(kuò)縮容。

Consumer 重啟不消費(fèi)：業(yè)務(wù)反饋消費(fèi)停止，重啟也不能夠解決問題，后來定位發(fā)現(xiàn)是早于 0.11 之前版本的 Bug：

https://issues.apache.org/jira/browse/KAFKA-5413 

原因是 log cleaner 線程掛了導(dǎo)致 Compact 停止，__consumer_offsets 這個(gè) Topic 的量非常大，broker reload 時(shí)間特別長(zhǎng)，這段時(shí)間是停止服務(wù)的。

解決方法有兩個(gè)：

• 一是升級(jí)到 Kafka 0.11+ 版本
• 二是將 Offset 遷移到新的 Consumer Group 來解決(規(guī)避掉有問題的 Coordinator)。

嚴(yán)鎖鵬

 

[[286313]] 

嚴(yán)鎖鵬，奇虎 360 大數(shù)據(jù)架構(gòu)運(yùn)維專家，具有 10 年基礎(chǔ)架構(gòu)與大數(shù)據(jù)開發(fā)經(jīng)驗(yàn)。2013 年加入 360 商業(yè)化團(tuán)隊(duì)，負(fù)責(zé)消息中間件開發(fā)與運(yùn)維，同時(shí)涉及大數(shù)據(jù)架構(gòu)、微服務(wù)架構(gòu)、實(shí)時(shí)計(jì)算平臺(tái)、機(jī)器學(xué)習(xí)平臺(tái)、監(jiān)控系統(tǒng)等基礎(chǔ)設(shè)施建設(shè)，致力于為商業(yè)化團(tuán)隊(duì)提供穩(wěn)定高效的基礎(chǔ)服務(wù)。