原創(chuàng)文章，轉(zhuǎn)載請注明出處
Kafka，近年來比較流行的一款分布式消息訂閱發(fā)布系統(tǒng)。由于其高吞吐量、可持久化以及分布式等特點(diǎn)得到了廣泛的應(yīng)用。最近部門搭建 Monitor 日志系統(tǒng)，需要在 Monitor Server 和產(chǎn)生日志的應(yīng)用中間放置一個(gè)消息隊(duì)列做為緩沖，在解耦應(yīng)用的同時(shí)提高 Monitor 系統(tǒng)的峰值處理能力，最終選定了使用 Kafka。在做系統(tǒng)的過程中學(xué)習(xí)到了不少Kafka的設(shè)計(jì)邏輯，對互聯(lián)網(wǎng)系統(tǒng)的認(rèn)識(shí)很有幫助。下面主要就 Kafka 的基本概念和配置進(jìn)行一些總結(jié)。
消息隊(duì)列
在目前的消息隊(duì)列系統(tǒng)中，有幾個(gè)重要的概念，可能在不同的系統(tǒng)中叫法不一樣，但是其在系統(tǒng)中的作用是互通的。
生產(chǎn)者(Producer)：消息的產(chǎn)生方，將“消息”這個(gè)實(shí)體發(fā)送到隊(duì)列中。
消費(fèi)者(Consumer)：消息的使用方，從隊(duì)列中獲取消息。
消息(Record)：生產(chǎn)和消費(fèi)的內(nèi)容，實(shí)際要傳輸?shù)臄?shù)據(jù)。
主題(Topic)：用于將消息進(jìn)行分類，生產(chǎn)者和消費(fèi)者可以指定 topic 進(jìn)行生產(chǎn)和消費(fèi)行為，可以將topic 理解為一個(gè)用于存放消息的隊(duì)列。
偏移量(Offset)：用于標(biāo)識(shí)被消費(fèi)的消息所在隊(duì)列中的位置。
除了這些，Kafka 中還有一些特有的概念：
消費(fèi)者組(Consumer Group)：將消費(fèi)者分組，維護(hù)其在不同 topic 之間的訂閱權(quán)限。
Broker：分布式系統(tǒng)中一個(gè)具體的服務(wù)，一個(gè) Kafka 集群一般由多個(gè) Broker 組成，以達(dá)到高可用的目的。
分區(qū)(Partition)：物理概念，在一個(gè) topic 內(nèi)部可以有若干個(gè) Partition，每個(gè) Partition 是實(shí)際儲(chǔ)存消息的隊(duì)列，若有多臺(tái)Broker，Partition會(huì)盡量被分配到不同的Broker上。生產(chǎn)者產(chǎn)生的消息只會(huì)進(jìn)入到其中一個(gè) Partiton 中。
Consumer Group
簡單來講，消息隊(duì)列模型可以分為兩種：基礎(chǔ)隊(duì)列模式和發(fā)布-訂閱模式。在基礎(chǔ)隊(duì)列模式中，一條消息只能由一個(gè)消費(fèi)者進(jìn)行處理，而在發(fā)布-訂閱模式中，一條消息可以分發(fā)給所有“訂閱”了此主題(Topic)的消費(fèi)者。Kafka 將這兩種模型合二為一，產(chǎn)生了消費(fèi)者組(Group)的概念。在不同的Group間，消息分發(fā)遵循發(fā)布-訂閱模式，即消息會(huì)被分發(fā)到所有訂閱了此Topic的 Group 中，而在 Group 內(nèi)部，消息分發(fā)遵循基礎(chǔ)隊(duì)列模式，即消息在 Group 中最終只會(huì)由一個(gè)消費(fèi)者消費(fèi)掉。
Partition
Kafka 不僅在 Consumer 層面做了分組控制，在 Topic 隊(duì)列中同樣存在著“分區(qū)”的概念。在基本模式中，一個(gè) Topic 就是一個(gè)隊(duì)列，這樣雖然非常清晰，但是在大數(shù)據(jù)量下，其性能直接受限于物理機(jī)。在分布式環(huán)境中，若采用簡單的“一個(gè)隊(duì)列”模型，會(huì)導(dǎo)致服務(wù)器壓力分布不均，假設(shè)服務(wù)器A(Broker A) 中的 Topic 消息壓力較大，而其他機(jī)器無法幫助其分擔(dān)，那么可能會(huì)直接導(dǎo)致服務(wù)器A(Broker A) 崩掉。
在 Kafka 的設(shè)計(jì)中，Topic 是邏輯概念，Partition 是物理概念，對于簡單的 Producer 和 Consumer ，兩者均不需要考慮 Partiton 的存在，只需要在 Topic 層面進(jìn)行操作即可。Partiton 的設(shè)計(jì)對于分布式系統(tǒng)有重要意義，Kafka 將一個(gè) Topic 打散成為若干 Partition，并且盡量保證 Partition 能平均分布到集群中的服務(wù)器(Broker)上。這樣也就解決了兩個(gè)重要問題— 負(fù)載均衡 和 水平擴(kuò)展。
Producer 產(chǎn)生的消息最終只會(huì)進(jìn)入到一個(gè) Partition 中，至于進(jìn)入哪個(gè) Partition，默認(rèn)的策略是平均分配，當(dāng)然在 Java 中我們也可以繼承Partitioner類來編寫自己的分配策略。
在 Patition 方面還涉及到分區(qū)備份以及 Leader 的選取，將在后面的擴(kuò)展中進(jìn)行總結(jié)。
場景
- 消息系統(tǒng)
Kafka最基本的使用方式，基于訂閱-發(fā)布的消息系統(tǒng)。和其他消息隊(duì)列產(chǎn)品相同，主要為系統(tǒng)帶來了異步處理和解耦的特性。
在這種場景下，Kafka是應(yīng)用之間聯(lián)系的紐帶，也是一個(gè)標(biāo)準(zhǔn)，有的文章將其比作人體的神經(jīng)系統(tǒng)。這是很準(zhǔn)確的，身體的許多器官之間之所以可以相互協(xié)作調(diào)用(最常見的就是大腦控制身體的運(yùn)動(dòng))，完全依靠神經(jīng)系統(tǒng)傳遞電信號(hào)。這些信號(hào)就是一個(gè)個(gè)消息，大腦產(chǎn)生這些信號(hào)，并將信號(hào)交給神經(jīng)系統(tǒng)讓它傳遞給具體的器官，器官接收到信號(hào)后根據(jù)信號(hào)的指令進(jìn)行具體的操作。
我們可以看一下消息系統(tǒng)的特點(diǎn)：
解耦。各個(gè)應(yīng)用之間更獨(dú)立，不必受其他應(yīng)用的影響。大家都通過一個(gè)簡單的消息隊(duì)列來聯(lián)系，不必受其他應(yīng)用返回值的影響，當(dāng)一個(gè)應(yīng)用掛掉之后其他應(yīng)用還是可以正常運(yùn)行
異步，提速。所謂提速，是指有大量并發(fā)的情況下，系統(tǒng)能夠盡快的做出回應(yīng)。如果是傳統(tǒng)的同步調(diào)用服務(wù)，在這項(xiàng)業(yè)務(wù)沒有完成之前，線程一直處于繁忙的狀態(tài)，當(dāng)并發(fā)量增加時(shí)，新的請求將會(huì)被阻塞，直至前面的線程處理完。更糟糕的情況是處于多個(gè)服務(wù)協(xié)作的系統(tǒng)中，當(dāng)有RPC請求時(shí)，當(dāng)前系統(tǒng)的線程將會(huì)處于阻塞狀態(tài)，對于分秒必爭的CPU資源，線程阻塞是非常昂貴的開銷。
通過消息系統(tǒng)，可以有效使CPU避免由于RPC引起的阻塞。當(dāng)應(yīng)用需要通過RPC調(diào)用其他服務(wù)時(shí)，可以直接向消息隊(duì)列發(fā)送一份信息，之后馬上返回繼續(xù)處理。這樣便可以將CPU的資源最大化利用。但是這需要架構(gòu)師考慮清楚幾點(diǎn)：
系統(tǒng)是否能夠容忍暫時(shí)的不一致性，或者說系統(tǒng)是否必須立馬知道RPC執(zhí)行的結(jié)果。顯然，引入消息系統(tǒng)，當(dāng)上層消息發(fā)送完消息并繼續(xù)往下執(zhí)行時(shí)，有一個(gè)潛在的邏輯是有下游系統(tǒng)會(huì)去做這件事， 但是實(shí)際做沒做，并不知道。
引入消息系統(tǒng)帶來的收益是否超過了引入的成本。引入成本自然不必說，新系統(tǒng)的布置維護(hù)等都是需要考慮的因素。如果盲目的照搬，“別人都這樣用，我們也這樣用”是架構(gòu)師的大忌，只有適合自己業(yè)務(wù)的架構(gòu)才是最好的架構(gòu)。
- 日志系統(tǒng)
其實(shí)日志系統(tǒng)使用Kafka也算消息隊(duì)列的一種應(yīng)用，只是由于日志系統(tǒng)的特性，使得消息隊(duì)列在這里格外重要。
對于日志系統(tǒng)來說，需要將若干應(yīng)用的日志統(tǒng)一搜集，然后進(jìn)行整理與存儲(chǔ)。日志系統(tǒng)的特點(diǎn)在于其量大、可能存在峰值，若使用單一的App向日志系統(tǒng)Push的模式，服務(wù)器的壓力可想而知。更關(guān)鍵的一點(diǎn)在于日志系統(tǒng)作為Monitor System的一部分，其本身的穩(wěn)定性要求就很高，一旦日志系統(tǒng)掛掉，可能會(huì)丟失大量的系統(tǒng)日志。
這時(shí)候Kafka(消息隊(duì)列)的有勢就顯現(xiàn)出來了：
第一，切斷了單個(gè)App與日志系統(tǒng)的直接聯(lián)系，解耦。因?yàn)槿罩镜乃鸭c分析實(shí)際上App本身并不很關(guān)心，將其發(fā)送到MQ就夠了，至于后續(xù)怎么處理，那App本身可以不管。
第二，削峰，有了MQ作為緩沖，當(dāng)不同應(yīng)用有大量Log需要傳輸時(shí)，數(shù)據(jù)有了合適的積壓點(diǎn)，在大流量下不至于將日志系統(tǒng)沖垮。而Kafka中的數(shù)據(jù)是以文件的形式落地在硬盤上的，保證了數(shù)據(jù)不會(huì)丟失，即便日志系統(tǒng)未來得及消費(fèi)，MQ中的消息也會(huì)保存一段時(shí)間(可配置，一般一周就可以了)。
第三，擴(kuò)展性。這一點(diǎn)是顯而易見的，現(xiàn)在微服務(wù)的概念非常流行，當(dāng)企業(yè)自身的服務(wù)越來越多，日志的量也會(huì)越來越大。得益于Kafka天生的集群特性，很容易可以添加Broker以拓展MQ的性能與容量。
上述的使用場景描述的還是比較籠統(tǒng)，實(shí)際業(yè)務(wù)中需要仔細(xì)考慮遇到的瓶頸在哪里，而引入MQ是否能夠解決這些問題，這需要對MQ的特點(diǎn)有詳細(xì)的了解。再者對于開發(fā)人員，不能一味照搬，無論在成本(開發(fā)成本，服務(wù)器成本，后期維護(hù)成本)還是使用方法上都需要慎重考慮。同時(shí)也是為自己提個(gè)醒，在使用一項(xiàng)技術(shù)之前一定要搞清楚為什么要用它。
簡單啟動(dòng)
了解完Kafka的基本概念之后我們可以簡單的上手操作一下，通過Kafka自帶的命令行工具和對一些參數(shù)的調(diào)整，直觀感受一下Kafka的運(yùn)作。
Kafka依賴zookeeper實(shí)現(xiàn)集群，所以我們先使用默認(rèn)的配置文件啟動(dòng)zk:
ZK啟動(dòng)之后，我們繼續(xù)使用默認(rèn)配置啟動(dòng)一個(gè)Kafka實(shí)例(Broker):
現(xiàn)在我們有一個(gè)運(yùn)行的zk實(shí)例和一個(gè)Kafka broker，它們已經(jīng)可以實(shí)現(xiàn)MQ的基本功能。接下來我們需要一個(gè)Producer和一個(gè)Consumer來進(jìn)行測試。Kafka的包中為我們提供了兩個(gè)命令行工具：kafka-console-producer和kafka-console-consumer，使得我們可以通過命令行來發(fā)送和消費(fèi)消息。
通過命令行啟動(dòng)Producer和Consumer，并發(fā)送一條測試消息：
這里為了演示使用的是官方提供的命令行工具，Kafka還提供了非常易用的Java API供開發(fā)者使用，能夠很方便的建立Producer和Consumer。
擴(kuò)展
Leader如何選舉？
首先何謂Leader？在之前的內(nèi)容中我們了解到了Kafka將整個(gè)Topic拆分為若干個(gè)Partition的作用，使每個(gè)Partition可以分布在不同的Broker上，以保證負(fù)載均衡。但是這樣還不夠，因?yàn)闊o法避免單點(diǎn)故障。
讀者可能會(huì)問：我這已經(jīng)將消息分布到不同的Partition中了，怎么還有單點(diǎn)故障呢？這里有一個(gè)非常重要的邏輯，那就是一條消息只能進(jìn)入一個(gè)Partition中，我們區(qū)分Partition、使其分布在不同的Broker上，只是為了保證更大的負(fù)載量和更有效的利用服務(wù)器資源。但是當(dāng)其中一個(gè)Broker突然掛掉了怎么辦呢？在這個(gè)Broker上的那些Partition中還有未被消費(fèi)的消息，如果沒有補(bǔ)救措施，它們就這樣丟失了嗎？
Kafka當(dāng)然考慮到了這點(diǎn)，所以有了副本和Leader的概念。
為了保證系統(tǒng)的高可用，每一個(gè)分區(qū)（Partition）都會(huì)有一定數(shù)量的副本（Replica），這樣如果部分服務(wù)器不可用，副本所在的服務(wù)器就會(huì)接管上來，保證應(yīng)用的持續(xù)性。同時(shí)為了保證較高的處理效率，消息的讀寫都是固定在同一個(gè)副本上的，這個(gè)副本就是所謂的Leader，其他作為備份存在的副本則被稱為Follower。值得注意的一點(diǎn)是在創(chuàng)建Topic配置參數(shù)時(shí)，副本（Replica）的數(shù)量不可以超過Broker的數(shù)量。
先來看一下 Leader 和 Follower 在 Broker 上的分布情況。
Kafka 使用分配算法來保證每個(gè) Broker 上的 Replica 數(shù)量盡量平均，從而使單個(gè) Broker 的壓力不會(huì)過大。關(guān)于詳細(xì)的分配算法我們可以從 Kafka 官方的文檔中查到，這里我們只需要簡單的理解為：
Kafka在分配 Partition 時(shí)會(huì)盡量平均的將 Partition 分配到不同的 Broker 中，然后從當(dāng)前 Broker 開始順序地為此 Partition 創(chuàng)建 Replica。
為了保證較高的處理效率，消息的讀寫都是由Leader來完成，那怎樣從眾多Follower中選舉出Leader就成了重要的話題。
這里不得不提到兩個(gè)概念：AR（Assigned Replicas）和ISR（In-Sync Replicas）。當(dāng)我們?yōu)橐粋€(gè)Topic設(shè)置了多個(gè)Replica的時(shí)候，這些所有的副本統(tǒng)稱為AR，AR中的Follower會(huì)不斷的從Leader中同步數(shù)據(jù)，同步的過程必然有所延遲，這是由于不同機(jī)器的性能等因素導(dǎo)致的。
ISR是AR的子集，它表示的是與Leader同步的Follower的集合。當(dāng)Follower從Leader同步，延遲超過了閾值，將會(huì)被Leader從ISR中踢出，Kafka 0.10.x中的參數(shù)replica.lag.time.max.ms，表示了延遲的閾值。
例如上圖中的 Partition 1只有一個(gè)Replica，R1，那么 R1就屬于 Partition 1的AR，如果 R1能在延遲閾值之內(nèi)保持與 Leader 的同步，那么 R1就屬于 ISR。
ISR這個(gè)集合非常重要，正是它保證了Kafka的高可用性。當(dāng)一條消息被Produce或者需要被Consume的時(shí)候，Leader需要等到ISR中的所有副本將本次操作的狀態(tài)同步，再返回出結(jié)果。這樣雖然在性能上有所下降，但是大大提高了Kafka的高可用性。
當(dāng)Leader掛掉，Kafka會(huì)從ISR中選取一個(gè)Follower作為新的Leader，因?yàn)镮SR中Follower的數(shù)據(jù)是與前任Leader同步的，所以其中任何一個(gè)Follower都可以直接充當(dāng)Leader的角色。
Kafka的高可用性有很多值得探討的地方，篇幅所限這里就不再深入討論，推薦一篇不錯(cuò)的文章，有興趣的朋友可以看看：
