Kafka 是當(dāng)下熱門(mén)的消息隊(duì)列中間件，它可以實(shí)時(shí)地處理海量數(shù)據(jù)，具備高吞吐、低延時(shí)等特性及可靠的消息異步傳遞機(jī)制，可以很好地解決不同系統(tǒng)間數(shù)據(jù)的交流和傳遞問(wèn)題。

Kafka 在馬蜂窩也有非常廣泛的應(yīng)用，為很多核心的業(yè)務(wù)提供支撐。本文將圍繞 Kafka 在馬蜂窩大數(shù)據(jù)平臺(tái)的應(yīng)用實(shí)踐，介紹相關(guān)業(yè)務(wù)場(chǎng)景、在 Kafka 應(yīng)用的不同階段我們遇到了哪些問(wèn)題以及如何解決、之后還有哪些計(jì)劃等。

Part.1.應(yīng)用場(chǎng)景

從 Kafka 在大數(shù)據(jù)平臺(tái)的應(yīng)用場(chǎng)景來(lái)看，主要分為以下三類：

第一類是將 Kafka 作為數(shù)據(jù)庫(kù)，提供大數(shù)據(jù)平臺(tái)對(duì)實(shí)時(shí)數(shù)據(jù)的存儲(chǔ)服務(wù)。從來(lái)源和用途兩個(gè)維度來(lái)說(shuō)，可以將實(shí)時(shí)數(shù)據(jù)分為業(yè)務(wù)端 DB 數(shù)據(jù)、監(jiān)控類型日志、基于埋點(diǎn)的客戶端日志(H5、WEB、APP、小程序)和服務(wù)端日志。

第二類是為數(shù)據(jù)分析提供數(shù)據(jù)源，各埋點(diǎn)日志會(huì)作為數(shù)據(jù)源，支持并對(duì)接公司離線數(shù)據(jù)、實(shí)時(shí)數(shù)據(jù)倉(cāng)庫(kù)及分析系統(tǒng)，包括多維查詢、實(shí)時(shí) Druid OLAP、日志明細(xì)等。

第三類是為業(yè)務(wù)方提供數(shù)據(jù)訂閱。除了在大數(shù)據(jù)平臺(tái)內(nèi)部的應(yīng)用之外，我們還使用 Kafka 為推薦搜索、大交通、酒店、內(nèi)容中心等核心業(yè)務(wù)提供數(shù)據(jù)訂閱服務(wù)，如用戶實(shí)時(shí)特征計(jì)算、用戶實(shí)時(shí)畫(huà)像訓(xùn)練及實(shí)時(shí)推薦、反作弊、業(yè)務(wù)監(jiān)控報(bào)警等。

主要應(yīng)用如下圖所示：

Part.2.演進(jìn)之路

四個(gè)階段

早期大數(shù)據(jù)平臺(tái)之所以引入 Kafka 作為業(yè)務(wù)日志的收集處理系統(tǒng)，主要是考慮到它高吞吐低延遲、多重訂閱、數(shù)據(jù)回溯等特點(diǎn)，可以更好地滿足大數(shù)據(jù)場(chǎng)景的需求。但隨著業(yè)務(wù)量的迅速增加，以及在業(yè)務(wù)使用和系統(tǒng)維護(hù)中遇到的問(wèn)題，例如注冊(cè)機(jī)制、監(jiān)控機(jī)制等的不完善，導(dǎo)致出現(xiàn)問(wèn)題無(wú)法快速定位，以及一些線上實(shí)時(shí)任務(wù)發(fā)生故障后沒(méi)有快速恢復(fù)導(dǎo)致消息積壓等, 使 Kafka 集群的穩(wěn)定性和可用性得受到挑戰(zhàn)，經(jīng)歷了幾次嚴(yán)重的故障。

解決以上問(wèn)題對(duì)我們來(lái)說(shuō)迫切而棘手。針對(duì)大數(shù)據(jù)平臺(tái)在使用 Kafka 上存在的一些痛點(diǎn)，我們從集群使用到應(yīng)用層擴(kuò)展做了一系列的實(shí)踐，整體來(lái)說(shuō)包括四個(gè)階段：

第一階段：版本升級(jí)。圍繞平臺(tái)數(shù)據(jù)生產(chǎn)和消費(fèi)方面存在的一些瓶頸和問(wèn)題，我們針對(duì)目前的 Kafka 版本進(jìn)行技術(shù)選型，最終確定使用 1.1.1 版本。

第二階段：資源隔離。為了支持業(yè)務(wù)的快速發(fā)展，我們完善了多集群建設(shè)以及集群內(nèi) Topic 間的資源隔離。

第三階段：權(quán)限控制和監(jiān)控告警。

首先在安全方面，早期的 Kafka 集群處于裸跑狀態(tài)。由于多產(chǎn)品線共用 Kafka，很容易由于誤讀其他業(yè)務(wù)的 Topic 導(dǎo)致數(shù)據(jù)安全問(wèn)題。因此我們基于 SASL/ SCRAM + ACL 增加了鑒權(quán)的功能。

在監(jiān)控告警方面，Kafka 目前已然成為實(shí)時(shí)計(jì)算中輸入數(shù)據(jù)源的標(biāo)配，那么其中 Lag 積壓情況、吞吐情況就成為實(shí)時(shí)任務(wù)是否健康的重要指標(biāo)。因此，大數(shù)據(jù)平臺(tái)構(gòu)建了統(tǒng)一的 Kafka 監(jiān)控告警平臺(tái)并命名「雷達(dá)」，多維度監(jiān)控 Kafka 集群及使用方情況。

第四階段：應(yīng)用擴(kuò)展。早期 Kafka 在對(duì)公司各業(yè)務(wù)線開(kāi)放的過(guò)程中，由于缺乏統(tǒng)一的使用規(guī)范，導(dǎo)致了一些業(yè)務(wù)方的不正確使用。為解決該痛點(diǎn)，我們構(gòu)建了實(shí)時(shí)訂閱平臺(tái)，通過(guò)應(yīng)用服務(wù)的形式賦能給業(yè)務(wù)方，實(shí)現(xiàn)數(shù)據(jù)生產(chǎn)和消費(fèi)申請(qǐng)、平臺(tái)的用戶授權(quán)、使用方監(jiān)控告警等眾多環(huán)節(jié)流程化自動(dòng)化，打造從需求方使用到資源全方位管控的整體閉環(huán)。

下面圍繞幾個(gè)關(guān)鍵點(diǎn)為大家展開(kāi)介紹。

核心實(shí)踐

1. 版本升級(jí)

之前大數(shù)據(jù)平臺(tái)一直使用的是 0.8.3 這一 Kafka 早期版本，而截止到當(dāng)前，Kafka 官方最新的 Release 版本已經(jīng)到了 2.3，于是長(zhǎng)期使用 0.8 版本過(guò)程中漸漸遇到的很多瓶頸和問(wèn)題，我們是能夠通過(guò)版本升級(jí)來(lái)解決的。

舉例來(lái)說(shuō)，以下是一些之前使用舊版時(shí)常見(jiàn)的問(wèn)題：

• 缺少對(duì) Security 的支持:存在數(shù)據(jù)安全性問(wèn)題及無(wú)法通過(guò)認(rèn)證授權(quán)對(duì)資源使用細(xì)粒度管理
• broker under replicated：發(fā)現(xiàn) broker 處于 under replicated 狀態(tài)，但不確定問(wèn)題的產(chǎn)生原因，難以解決。
• 新的 feature 無(wú)法使用：如事務(wù)消息、冪等消息、消息時(shí)間戳、消息查詢等。
• 客戶端的對(duì) offset 的管理依賴 zookeeper, 對(duì) zookeeper 的使用過(guò)重, 增加運(yùn)維的復(fù)雜度
• 監(jiān)控指標(biāo)不完善：如 topic、partition、broker 的數(shù)據(jù) size 指標(biāo), 同時(shí) kafka manager 等監(jiān)控工具對(duì)低版本 kafka 支持不好

同時(shí)對(duì)一些目標(biāo)版本的特性進(jìn)行了選型調(diào)研，如：

• 0.9 版本, 增加了配額和安全性, 其中安全認(rèn)證和授權(quán)是我們最關(guān)注的功能
• 0.10 版本，更細(xì)粒度的時(shí)間戳. 可以基于偏移量進(jìn)行快速的數(shù)據(jù)查找，找到所要的時(shí)間戳。這在實(shí)時(shí)數(shù)據(jù)處理中基于 Kafka 數(shù)據(jù)源的數(shù)據(jù)重播是極其重要的
• 0.11 版本, 冪等性和 Transactions 的支持及副本數(shù)據(jù)丟失/數(shù)據(jù)不一致的解決。
  • 冪等性意味著對(duì)于同一個(gè) Partition，面對(duì) Data 的多次發(fā)布，Kafka broker 端就可以做到自動(dòng)去重;
  • 對(duì) Transactions 的支持使一個(gè)事務(wù)下發(fā)布多條信息到多個(gè) Topic Partition 時(shí)，我們可以使它以原子性的方式被完成。在我們的下游消費(fèi)者中，很多都是用 Flink 做一些流處理的工作，因此在數(shù)據(jù)處理及故障恢復(fù)時(shí)僅一次語(yǔ)義則顯得尤為重要。而 0.11 版本對(duì)于事務(wù)的支持則可以保證與 Kafka 交互的 Flink 應(yīng)用實(shí)現(xiàn)端到端僅一次語(yǔ)義, 支持 EOS 可以對(duì)數(shù)據(jù)可靠性有絕對(duì)要求, 比如交易、風(fēng)控等場(chǎng)景下的重要支持。
  • Leader Epoch：解決了原先依賴水位表示副本進(jìn)度可能造成的數(shù)據(jù)丟失/數(shù)據(jù)不一致問(wèn)題。
• 1.1 版本，運(yùn)維性的提升。比如當(dāng) Controller Shut Down，想要關(guān)閉一個(gè) Broker 的時(shí)候，之前需要一個(gè)很長(zhǎng)很復(fù)雜的過(guò)程在 1.0 版本得到很大的改善。

最終選擇 1.1 版本, 則是因?yàn)槌鲇?Camus 與 Kafka 版本的兼容性及 1.1 版本已經(jīng)滿足了使用場(chǎng)景中重要新特性的支持的綜合考量。這里再簡(jiǎn)單說(shuō)一下 Camus 組件，同樣是由 Linkedin 開(kāi)源，在我們的大數(shù)據(jù)平臺(tái)中主要作為 Kafka 數(shù)據(jù) Dump 到 HDFS 的重要方式。

2. 資源隔離

之前由于業(yè)務(wù)的復(fù)雜性和規(guī)模不大，大數(shù)據(jù)平臺(tái)對(duì)于 Kafka 集群的劃分比較簡(jiǎn)單。于是，一段時(shí)間以后導(dǎo)致公司業(yè)務(wù)數(shù)據(jù)混雜在一起，某一個(gè)業(yè)務(wù)主題存在的不合理使用都有可能導(dǎo)致某些 Broker 負(fù)載過(guò)重，影響到其他正常的業(yè)務(wù)，甚至某些 Broker 的故障會(huì)出現(xiàn)影響整個(gè)集群，導(dǎo)致全公司業(yè)務(wù)不可用的風(fēng)險(xiǎn)。

針對(duì)以上的問(wèn)題，在集群改造上做了兩方面實(shí)踐

按功能屬性拆分獨(dú)立的集群

集群內(nèi)部 Topic 粒度的資源隔離

(1)集群拆分

按照功能維度拆分多個(gè) Kafka 物理集群，進(jìn)行業(yè)務(wù)隔離，降低運(yùn)維復(fù)雜度。

以目前最重要的埋點(diǎn)數(shù)據(jù)使用來(lái)說(shuō), 目前拆分為三類集群，各類集群的功能定義如下：

• Log 集群：各端的埋點(diǎn)數(shù)據(jù)采集后會(huì)優(yōu)先落地到該集群, 所以這個(gè)過(guò)程不能出現(xiàn)由于 Kafka 問(wèn)題導(dǎo)致采集中斷，這對(duì) Kafka 可用性要求很高。因此該集群不會(huì)對(duì)外提供訂閱，保證消費(fèi)方可控;同時(shí)該集群業(yè)務(wù)也作為離線采集的源頭，數(shù)據(jù)會(huì)通過(guò) Camus 組件按小時(shí)時(shí)間粒度 dump 到 HDFS 中，這部分?jǐn)?shù)據(jù)參與后續(xù)的離線計(jì)算。
• 全量訂閱集群：該集群 Topic 中的絕大部分?jǐn)?shù)據(jù)是從 Log 集群實(shí)時(shí)同步過(guò)來(lái)的。上面我們提到了 Log 集群的數(shù)據(jù)是不對(duì)外的，因此全量集群就承擔(dān)了消費(fèi)訂閱的職責(zé)。目前主要是用于平臺(tái)內(nèi)部的實(shí)時(shí)任務(wù)中，來(lái)對(duì)多個(gè)業(yè)務(wù)線的數(shù)據(jù)分析并提供分析服務(wù)。
• 個(gè)性定制集群：之前提到過(guò)，我們可以根據(jù)業(yè)務(wù)方需求來(lái)拆分、合并數(shù)據(jù)日志源，同時(shí)我們還支持定制化 Topic，該集群只需要提供分流后 Topic 的落地存儲(chǔ)。

集群整體架構(gòu)劃分如下圖：

(2)資源隔離

Topic 的流量大小是集群內(nèi)部進(jìn)行資源隔離的重要依據(jù)。例如，我們?cè)跇I(yè)務(wù)中埋點(diǎn)日志量較大的兩個(gè)數(shù)據(jù)源分別是后端埋點(diǎn)數(shù)據(jù)源 server-event 和端上的埋點(diǎn) mobile-event 數(shù)據(jù)源，我們要避免存儲(chǔ)兩個(gè)數(shù)據(jù)的主題分區(qū)分配到集群中同一個(gè) Broker 上的節(jié)點(diǎn)。通過(guò)在不同 Topic 進(jìn)行物理隔離，就可以避免 Broker 上的流量發(fā)生傾斜。

3. 權(quán)限控制和監(jiān)控告警

(1)權(quán)限控制

開(kāi)始介紹時(shí)我們說(shuō)過(guò)，早期 Kafka 集群沒(méi)有設(shè)置安全驗(yàn)證處于裸跑狀態(tài)，因此只要知道 Broker 的連接地址即可生產(chǎn)消費(fèi)，存在嚴(yán)重的數(shù)據(jù)安全性問(wèn)題。

一般來(lái)說(shuō), 使用 SASL 的用戶多會(huì)選擇 Kerberos，但就平臺(tái) Kafka 集群的使用場(chǎng)景來(lái)說(shuō)，用戶系統(tǒng)并不復(fù)雜，使用 Kerberos 就有些大材小用, 同時(shí) Kerberos 相對(duì)復(fù)雜，存在引發(fā)其他問(wèn)題的風(fēng)險(xiǎn)。另外，在 Encryption 方面, 由于都是運(yùn)行在內(nèi)網(wǎng)環(huán)境，所以并沒(méi)有使用 SSL 加密。

最終平臺(tái) Kafka 集群使用 SASL 作為鑒權(quán)方式, 基于 SASL/ SCRAM + ACL 的輕量級(jí)組合方式，實(shí)現(xiàn)動(dòng)態(tài)創(chuàng)建用戶，保障數(shù)據(jù)安全。

(2)監(jiān)控告警

之前在集群的使用中我們經(jīng)常發(fā)現(xiàn)，消費(fèi)應(yīng)用的性能無(wú)緣無(wú)故變差了。分析問(wèn)題的原因, 通常是滯后 Consumer 讀取的數(shù)據(jù)大概率沒(méi)有命中 Page- cache，導(dǎo)致 Broker 端機(jī)器的內(nèi)核要首先從磁盤(pán)讀取數(shù)據(jù)加載到 Page- cache 中后，才能將結(jié)果返還給 Consumer，相當(dāng)于本來(lái)可以服務(wù)于寫(xiě)操作的磁盤(pán)現(xiàn)在要讀取數(shù)據(jù)了, 影響了使用方讀寫(xiě)同時(shí)降低的集群的性能。

這時(shí)就需要找出滯后 Consumer 的應(yīng)用進(jìn)行事前的干預(yù)從而減少問(wèn)題發(fā)生，因此監(jiān)控告警無(wú)論對(duì)平臺(tái)還是用戶都有著重大的意義。下面介紹一下我們的實(shí)踐思路。

整體方案：

整體方案主要是基于開(kāi)源組件 Kafka JMX Metrics+OpenFalcon+Grafana：

• Kafka JMX Metrics：Kafka broker 的內(nèi)部指標(biāo)都以 JMX Metrics 的形式暴露給外部。1.1.1 版本 提供了豐富的監(jiān)控指標(biāo)，滿足監(jiān)控需要
• OpenFalcon：小米開(kāi)源的一款企業(yè)級(jí)、高可用、可擴(kuò)展的開(kāi)源監(jiān)控系統(tǒng)
• Grafana：Metrics 可視化系統(tǒng)，大家比較熟悉，可對(duì)接多種 Metrics 數(shù)據(jù)源。

關(guān)于監(jiān)控：

• Falcon-agent：部署到每臺(tái) Broker 上, 解析 Kafka JMX 指標(biāo)上報(bào)數(shù)據(jù)
• Grafana：用來(lái)可視化 Falcon Kafka Metrics 數(shù)據(jù)，對(duì) Cluster、Broker、Topic、Consumer 4 個(gè)角色制作監(jiān)控大盤(pán)。
• Eagle：獲取消費(fèi)組 Active 狀態(tài)、消費(fèi)組 Lag 積壓情況，同時(shí)提供 API，為監(jiān)控告警系統(tǒng)「雷達(dá)」提供監(jiān)控?cái)?shù)據(jù)。

關(guān)于告警：

雷達(dá)系統(tǒng): 自研監(jiān)控系統(tǒng)，通過(guò) Falcon 及 Eagle 獲取 Kafka 指標(biāo)，結(jié)合設(shè)定閾值進(jìn)行告警。以消費(fèi)方式舉例，Lag 是衡量消費(fèi)情況是否正常的一個(gè)重要指標(biāo)，如果 Lag 一直增加，必須要對(duì)它進(jìn)行處理。

發(fā)生問(wèn)題的時(shí)候，不僅 Consumer 管理員要知道，它的用戶也要知道，所以報(bào)警系統(tǒng)也需要通知到用戶。具體方式是通過(guò)企業(yè)微信告警機(jī)器人自動(dòng)提醒對(duì)應(yīng)消費(fèi)組的負(fù)責(zé)人或使用者及 Kafka 集群的管理者。

監(jiān)控示例：

4. 應(yīng)用擴(kuò)展

(1)實(shí)時(shí)數(shù)據(jù)訂閱平臺(tái)

實(shí)時(shí)數(shù)據(jù)訂閱平臺(tái)是一個(gè)提供 Kafka 使用全流程管理的系統(tǒng)應(yīng)用，以工單審批的方式將數(shù)據(jù)生產(chǎn)和消費(fèi)申請(qǐng)、平臺(tái)用戶授權(quán)、使用方監(jiān)控告警等眾多環(huán)節(jié)流程化自動(dòng)化, 并提供統(tǒng)一管控。

核心思想是基于 Kafka 數(shù)據(jù)源的身份認(rèn)證和權(quán)限控制，增加數(shù)據(jù)安全性的同時(shí)對(duì) Kafka 下游應(yīng)用進(jìn)行管理。

(2)標(biāo)準(zhǔn)化的申請(qǐng)流程

無(wú)論生產(chǎn)者還是消費(fèi)者的需求，使用方首先會(huì)以工單的方式提出訂閱申請(qǐng)。申請(qǐng)信息包括業(yè)務(wù)線、Topic、訂閱方式等信息;工單最終會(huì)流轉(zhuǎn)到平臺(tái)等待審批;如果審批通過(guò)，使用方會(huì)分配到授權(quán)賬號(hào)及 Broker 地址。至此，使用方就可以進(jìn)行正常的生產(chǎn)消費(fèi)了。

(3)監(jiān)控告警

對(duì)于平臺(tái)來(lái)說(shuō)，權(quán)限與資源是綁定的，資源可以是用于生產(chǎn)的 Topic 或消費(fèi)使用的 GroupTopic。一旦權(quán)限分配后，對(duì)于該部分資源的使用就會(huì)自動(dòng)在我們的雷達(dá)監(jiān)控系統(tǒng)進(jìn)行注冊(cè)，用于資源整個(gè)生命的周期的監(jiān)控。

(4)數(shù)據(jù)重播

出于對(duì)數(shù)據(jù)完整性和準(zhǔn)確性的考量，目前 Lamda 架構(gòu)已經(jīng)是大數(shù)據(jù)的一種常用架構(gòu)方式。但從另一方面來(lái)說(shuō)， Lamda 架構(gòu)也存在資源的過(guò)多使用和開(kāi)發(fā)難度高等問(wèn)題。

實(shí)時(shí)訂閱平臺(tái)可以為消費(fèi)組提供任意位點(diǎn)的重置，支持對(duì)實(shí)時(shí)數(shù)據(jù)按時(shí)間、位點(diǎn)等多種方式的數(shù)據(jù)重播, 并提供對(duì) Kappa 架構(gòu)場(chǎng)景的支持，來(lái)解決以上痛點(diǎn)。

(5)主題管理

為什么提供主題管理?舉一些很簡(jiǎn)單的例子，比如當(dāng)我們想讓一個(gè)用戶在集群上創(chuàng)建他自己的 Kafka Topic，這時(shí)顯然是不希望讓他直接到一個(gè)節(jié)點(diǎn)上操作的。因此剛才所講的服務(wù)，不管是對(duì)用戶來(lái)講，還是管理員來(lái)講，我們都需要有一個(gè)界面操作它，因?yàn)椴豢赡芩腥硕纪ㄟ^(guò) SSH 去連服務(wù)器。

因此需要一個(gè)提供管理功能的服務(wù)，創(chuàng)建統(tǒng)一的入口并引入主題管理的服務(wù)，包括主題的創(chuàng)建、資源隔離指定、主題元數(shù)據(jù)管理等。

(6)數(shù)據(jù)分流

在之前的架構(gòu)中, 使用方消費(fèi) Kafka 數(shù)據(jù)的粒度都是每個(gè) Kafka Topic 保存 LogSource 的全量數(shù)據(jù)，但在使用中很多消費(fèi)方只需要消費(fèi)各 LogSource 的部分?jǐn)?shù)據(jù)，可能也就是某一個(gè)應(yīng)用下幾個(gè)埋點(diǎn)事件的數(shù)據(jù)。如果需要下游應(yīng)用自己寫(xiě)過(guò)濾規(guī)則，肯定存在資源的浪費(fèi)及使用便捷性的問(wèn)題;另外還有一部分場(chǎng)景是需要多個(gè)數(shù)據(jù)源 Merge 在一起來(lái)使用的。

基于上面的兩種情況, 我人實(shí)現(xiàn)了按業(yè)務(wù)方需求拆分、合并并定制化 Topic 支持跨數(shù)據(jù)源的數(shù)據(jù)合并及 appcode 和 event code 的任意組個(gè)條件的過(guò)濾規(guī)則。

Part.3.后續(xù)計(jì)劃

解決數(shù)據(jù)重復(fù)問(wèn)題。為了解決目前平臺(tái)實(shí)時(shí)流處理中因故障恢復(fù)等因素導(dǎo)致數(shù)據(jù)重復(fù)的問(wèn)題，我們正在嘗試用 Kafka 的事務(wù)機(jī)制結(jié)合 Flink 的兩段提交協(xié)議實(shí)現(xiàn)端到端的僅一次語(yǔ)義。目前已經(jīng)在平臺(tái)上小范圍試用, 如果通過(guò)測(cè)試，將會(huì)在生產(chǎn)環(huán)境下推廣。

Consumer 限流。在一寫(xiě)多讀場(chǎng)景中, 如果某一個(gè) Consumer 操作大量讀磁盤(pán), 會(huì)影響 Produce 級(jí)其他消費(fèi)者操作的延遲。l因此，通過(guò) Kafka Quota 機(jī)制對(duì) Consume 限流及支持動(dòng)態(tài)調(diào)整閾值也是我們后續(xù)的方向

場(chǎng)景擴(kuò)展?；?Kafka 擴(kuò)展 SDK、HTTP 等多種消息訂閱及生產(chǎn)方式，滿足不同語(yǔ)言環(huán)境及場(chǎng)景的使用需求。

【本文是51CTO專欄作者馬蜂窩技術(shù)的原創(chuàng)文章，作者微信公眾號(hào)馬蜂窩技術(shù)(ID：mfwtech)】

戳這里，看該作者更多好文