kafka 由 LinkedIn 公司推出的一個(gè)高吞吐的分布式消息系統(tǒng)，通俗的說(shuō)就是一個(gè)基于發(fā)布和訂閱的消息隊(duì)列。

官網(wǎng)地址：https://kafka.apache.org/intro

應(yīng)用場(chǎng)景

•  異步解構(gòu)：在上下游沒(méi)有強(qiáng)依賴的業(yè)務(wù)關(guān)系或針對(duì)單次請(qǐng)求不需要立刻處理的業(yè)務(wù)；
•  系統(tǒng)緩沖：有利于解決服務(wù)系統(tǒng)的吞吐量不一致的情況，尤其對(duì)處理速度較慢的服務(wù)來(lái)說(shuō)起到緩沖作用；
•  消峰作用：對(duì)于短時(shí)間偶現(xiàn)的極端流量，對(duì)后端的服務(wù)可以啟動(dòng)保護(hù)作用；
•   數(shù)據(jù)流處理：集成 spark 做實(shí)時(shí)數(shù)據(jù)流處理。

Kafka 拓?fù)鋱D（多副本機(jī)制）

由上圖我們可以發(fā)現(xiàn) Kafka 是分布式，同時(shí)對(duì)于每一個(gè)分區(qū)都存在多副本，同時(shí)整個(gè)集群的管理都通過(guò) zookeeper 管理。

Kafka 核心組件

broker

Kafka 服務(wù)器，負(fù)責(zé)消息存儲(chǔ)和轉(zhuǎn)發(fā)；一 broker 就代表一個(gè) kafka 節(jié)點(diǎn)。一個(gè) broker 可以包含多個(gè) topic

topic

消息類別，Kafka 按照 topic 來(lái)分類消息

partition

•  topic 的分區(qū)，一個(gè) topic 可以包含多個(gè) partition，topic 消息保存在各個(gè) partition 上；由于一個(gè) topic 能被分到多個(gè)分區(qū)上，給 kafka 提供給了并行的處理能力，這也正是 kafka 高吞吐的原因之一。
•  partition 物理上由多個(gè) segment 文件組成，每個(gè) segment 大小相等，順序讀寫（這也是 kafka 比較快的原因之一，不需要隨機(jī)寫）。每個(gè) Segment 數(shù)據(jù)文件以該段中最小的 offset ，文件擴(kuò)展名為.log。當(dāng)查找 offset 的 Message 的時(shí)候，通過(guò)二分查找快速找到 Message 所處于的 Segment 中。

offset

•  消息在日志中的位置，可以理解是消息在 partition 上的偏移量，也是代表該消息的 唯一序號(hào)。
•  同時(shí)也是主從之間的需要同步的信息。

Producer

生產(chǎn)者，負(fù)責(zé)向 Kafka Broker 發(fā)消息的客戶端

Consumer

消息消者，負(fù)責(zé)消費(fèi) Kafka Broker 中的消息

Consumer Group

消費(fèi)者組，每個(gè) Consumer 必須屬于一個(gè) group；（注意的是 一個(gè)分區(qū)只能由組內(nèi)一個(gè)消費(fèi)者消費(fèi)，消費(fèi)者組之間互不影響。）

Zookeeper

管理 kafka 集群，負(fù)責(zé)存儲(chǔ)了集群 broker、topic、partition 等 meta 數(shù)據(jù)存儲(chǔ)，同時(shí)也負(fù)責(zé) broker 故障發(fā)現(xiàn)，partition leader 選舉，負(fù)載均衡等功能。

服務(wù)治理

既然 Kafka 是分布式的發(fā)布/訂閱系統(tǒng)，這樣如果做的集群之間數(shù)據(jù)同步和一致性，kafka 是不是肯定不會(huì)丟消息呢？以及宕機(jī)的時(shí)候如果進(jìn)行 Leader 選舉呢？

數(shù)據(jù)同步

在 Kafka 中的 Partition 有一個(gè) leader 與多個(gè) follower，producer 往某個(gè) Partition 中寫入數(shù)據(jù)時(shí)，只會(huì)往 leader 中寫入數(shù)據(jù)，然后數(shù)據(jù)才會(huì)被復(fù)制進(jìn)其他的 Replica 中。而每一個(gè) follower 可以理解成一個(gè)消費(fèi)者，定期去 leader 去拉取消息。而只有數(shù)據(jù)同步了后，kafka 才會(huì)給生產(chǎn)者返回一個(gè) ACK 告知消息已經(jīng)存儲(chǔ)落地了。

ISR

在 Kafka 中，為了保證性能，Kafka 不會(huì)采用強(qiáng)一致性的方式來(lái)同步主從的數(shù)據(jù)。而是維護(hù)了一個(gè)：in-sync Replica 的列表，Leader 不需要等待所有 Follower 都完成同步，只要在 ISR 中的 Follower 完成數(shù)據(jù)同步就可以發(fā)送 ack 給生產(chǎn)者即可認(rèn)為消息同步完成。同時(shí)如果發(fā)現(xiàn) ISR 里面某一個(gè) follower 落后太多的話，就會(huì)把它剔除。

具體流程如下：

上述的做法并無(wú)法保證 kafka 一定不丟消息。 雖然 Kafka 通過(guò)多副本機(jī)制中最大限度保證消息不會(huì)丟失，但是如果數(shù)據(jù)已經(jīng)寫入系統(tǒng) page cache 中但是還沒(méi)來(lái)得及刷入磁盤，此時(shí)突然機(jī)器宕機(jī)或者掉電，那消息自然而然的就會(huì)丟失。

Kafka 故障恢復(fù)

 

Kafka 通過(guò) Zookeeper 連坐集群的管理，所以這里的選舉機(jī)制采用的是 Zab(zookeeper 使用)。

•  生產(chǎn)者發(fā)生消息給 leader，這個(gè)時(shí)候 leader 完成數(shù)據(jù)存儲(chǔ)，突然發(fā)生故障，沒(méi)有給 producer 返回 ack；
•  通過(guò) ZK 選舉，其中一個(gè) follower 成為 leader，這個(gè)時(shí)候 producer 重新請(qǐng)求新的 leader，并存儲(chǔ)數(shù)據(jù)。

Kafka 為什么這么快

順序?qū)懘疟P

Kafka 采用了順序?qū)懘疟P，而由于順序?qū)懘疟P相對(duì)隨機(jī)寫，減少了尋地址的耗費(fèi)時(shí)間。（在 Kafka 的每一個(gè)分區(qū)里面消息是有序的。

Page Cache

Kafka 在 OS 系統(tǒng)方面使用了 Page Cache 而不是我們平常所用的 Buffer。Page Cache 其實(shí)不陌生，也不是什么新鮮事物。

我們?cè)趌inux上查看內(nèi)存的時(shí)候，經(jīng)常可以看到buff/cache，兩者都是用來(lái)加速IO讀寫用的，而cache是作用于讀，也就是說(shuō)，磁盤的內(nèi)容可以讀到cache里面這樣，應(yīng)用程序讀磁盤就非?？?；而buff是作用于寫，我們開發(fā)寫磁盤都是，一般如果寫入一個(gè)buff里面再flush就非常快。而kafka正是把這兩者發(fā)揮了極致：Kafka雖然是scala寫的，但是依舊在Java的虛擬機(jī)上運(yùn)行，盡管如此，kafka它還是盡量避開了JVM的限制，它利用了Page cache來(lái)存儲(chǔ)，這樣躲開了數(shù)據(jù)在JVM因?yàn)镚C而發(fā)生的STW。另一方面也是Page Cache使得它實(shí)現(xiàn)了零拷貝，具體下面會(huì)講。

零拷貝

無(wú)論是優(yōu)秀的 Netty 還是其他優(yōu)秀的 Java 框架，基本都在零拷貝減少了 CPU 的上下文切換和磁盤的 IO。當(dāng)然 Kafka 也不例外。零拷貝的概念具體這里不作太詳細(xì)的復(fù)述，大致的給大家講一下這個(gè)概念。

傳統(tǒng)的一次應(yīng)用程序請(qǐng)求數(shù)據(jù)的過(guò)程

這里大致可以發(fā)傳統(tǒng)的方式發(fā)生了 4 次拷貝，2 次 DMA 和 2 次 CPU，而 CPU 發(fā)生了 4 次的切換。_（DMA 簡(jiǎn)單理解就是，在進(jìn)行 I/O 設(shè)備和內(nèi)存的數(shù)據(jù)傳輸?shù)臅r(shí)候，數(shù)據(jù)搬運(yùn)的工作全部交給 DMA 控制器，而 CPU 不再參與任何與數(shù)據(jù)搬運(yùn)相關(guān)的事情）。

零拷貝的方式

通過(guò)優(yōu)化我們可以發(fā)現(xiàn)，CPU 只發(fā)生了 2 次的上下文切換和 3 次數(shù)據(jù)拷貝。（linux 系統(tǒng)提供了系統(tǒng)事故調(diào)用函數(shù)“ sendfile()”，這樣系統(tǒng)調(diào)用，可以直接把內(nèi)核緩沖區(qū)里的數(shù)據(jù)拷貝到 socket 緩沖區(qū)里，不再拷貝到用戶態(tài)）。

分區(qū)分段

我們上面也介紹過(guò)了，kafka 采取了分區(qū)的模式，而每一個(gè)分區(qū)又對(duì)應(yīng)到一個(gè)物理分段，而查找的時(shí)候可以根據(jù)二分查找快速定位。這樣不僅提高了數(shù)據(jù)讀的查詢效率，也提供了并行操作的方式。

數(shù)據(jù)壓縮

Kafka 對(duì)數(shù)據(jù)提供了：Gzip 和 Snappy 壓縮協(xié)議等壓縮協(xié)議，對(duì)消息結(jié)構(gòu)體進(jìn)行了壓縮，一方面減少了帶寬，也減少了數(shù)據(jù)傳輸?shù)南摹?/p>

Kafka 安裝

安裝 JDK

由于使用壓縮包還需要自己配置環(huán)境變量，所以這里推薦直接用 yum 安裝，熟悉查看目前 Java 的版本：

$ yum -y list Java* 

安裝你想要的版本，這里我是 1.8 

$ yum install java-1.8.0-openjdk-devel.x86_64 

查看是否安裝成功 

$ Java -version 

安裝 Zookeeper

首先需要去官網(wǎng)下載安裝包，然后解壓 

$ tar -zxvf zookeeper-3.4.9.tar.gz 

要做的就是將這個(gè)文件復(fù)制一份，并命名為：zoo.cfg，然后在 zoo.cfg 中修改自己的配置即可 

$ cp zoo_sample.cfg zoo.cfg  
$ vim zoo.cfg 

主要配置解釋如下 

# zookeeper內(nèi)部的基本單位，單位是毫秒，這個(gè)表示一個(gè)tickTime為2000毫秒，在zookeeper的其他配置中，都是基于tickTime來(lái)做換算的  
tickTime=2000  
# 集群中的follower服務(wù)器(F)與leader服務(wù)器(L)之間 初始連接 時(shí)能容忍的最多心跳數(shù)（tickTime的數(shù)量）。  
initLimit=10  
#syncLimit：集群中的follower服務(wù)器(F)與leader服務(wù)器(L)之間 請(qǐng)求和應(yīng)答 之間能容忍的最多心跳數(shù)（tickTime的數(shù)量）  
syncLimit=5  
# 數(shù)據(jù)存放文件夾，zookeeper運(yùn)行過(guò)程中有兩個(gè)數(shù)據(jù)需要存儲(chǔ)，一個(gè)是快照數(shù)據(jù)（持久化數(shù)據(jù)）另一個(gè)是事務(wù)日志  
dataDir=/tmp/zookeeper  
## 客戶端訪問(wèn)端口  
clientPort=2181 

配置環(huán)境變量 

$ vim ~/.bash_profile  
$ export ZK=/usr/local/src/apache-zookeeper-3.7.0-bin  
$ export PATH=$PATH:$ZK/bin  
$ export PATH  
# 啟動(dòng)  
$ zkServer.sh start 

下面能看啟動(dòng)成功

安裝 Kafka

•  下載 kafka 

$ wget https://www.apache.org/dyn/closer.cgi?path=/kafka/2.8.0/kafka-2.8.0-src.tgz 

•  安裝 kafka 

$ tar -xzvf kafka_2.12-2.0.0.tgz 

•  配置環(huán)境變量 

$ export ZK=/usr/local/src/apache-zookeeper-3.7.0-bin  
$ export PATH=$PATH:$ZK/bin  
$ export KAFKA=/usr/local/src/kafka  
$ export PATH=$PATH:$KAFKA/bin 

•  啟動(dòng) Kafka 

$ nohup kafka-server-start.sh 自己的配置文件路徑/server.properties & 

大功告成！