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

官網地址：https://kafka.apache.org/intro

應用場景

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

Kafka 拓撲圖（多副本機制）

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

Kafka 核心組件

broker

Kafka 服務器，負責消息存儲和轉發(fā)；一 broker 就代表一個 kafka 節(jié)點。一個 broker 可以包含多個 topic

topic

消息類別，Kafka 按照 topic 來分類消息

partition

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

offset

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

Producer

生產者，負責向 Kafka Broker 發(fā)消息的客戶端

Consumer

消息消者，負責消費 Kafka Broker 中的消息

Consumer Group

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

Zookeeper

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

服務治理

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

數據同步

在 Kafka 中的 Partition 有一個 leader 與多個 follower，producer 往某個 Partition 中寫入數據時，只會往 leader 中寫入數據，然后數據才會被復制進其他的 Replica 中。而每一個 follower 可以理解成一個消費者，定期去 leader 去拉取消息。而只有數據同步了后，kafka 才會給生產者返回一個 ACK 告知消息已經存儲落地了。

ISR

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

具體流程如下：

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

Kafka 故障恢復

 

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

•  生產者發(fā)生消息給 leader，這個時候 leader 完成數據存儲，突然發(fā)生故障，沒有給 producer 返回 ack；
•  通過 ZK 選舉，其中一個 follower 成為 leader，這個時候 producer 重新請求新的 leader，并存儲數據。

Kafka 為什么這么快

順序寫磁盤

Kafka 采用了順序寫磁盤，而由于順序寫磁盤相對隨機寫，減少了尋地址的耗費時間。（在 Kafka 的每一個分區(qū)里面消息是有序的。

Page Cache

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

我們在linux上查看內存的時候，經?？梢钥吹絙uff/cache，兩者都是用來加速IO讀寫用的，而cache是作用于讀，也就是說，磁盤的內容可以讀到cache里面這樣，應用程序讀磁盤就非?？欤欢鴅uff是作用于寫，我們開發(fā)寫磁盤都是，一般如果寫入一個buff里面再flush就非?？?。而kafka正是把這兩者發(fā)揮了極致：Kafka雖然是scala寫的，但是依舊在Java的虛擬機上運行，盡管如此，kafka它還是盡量避開了JVM的限制，它利用了Page cache來存儲，這樣躲開了數據在JVM因為GC而發(fā)生的STW。另一方面也是Page Cache使得它實現(xiàn)了零拷貝，具體下面會講。

零拷貝

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

傳統(tǒng)的一次應用程序請求數據的過程

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

零拷貝的方式

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

分區(qū)分段

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

數據壓縮

Kafka 對數據提供了：Gzip 和 Snappy 壓縮協(xié)議等壓縮協(xié)議，對消息結構體進行了壓縮，一方面減少了帶寬，也減少了數據傳輸的消耗。

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

首先需要去官網下載安裝包，然后解壓 

$ tar -zxvf zookeeper-3.4.9.tar.gz 

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

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

主要配置解釋如下 

# zookeeper內部的基本單位，單位是毫秒，這個表示一個tickTime為2000毫秒，在zookeeper的其他配置中，都是基于tickTime來做換算的  
tickTime=2000  
# 集群中的follower服務器(F)與leader服務器(L)之間 初始連接 時能容忍的最多心跳數（tickTime的數量）。  
initLimit=10  
#syncLimit：集群中的follower服務器(F)與leader服務器(L)之間 請求和應答 之間能容忍的最多心跳數（tickTime的數量）  
syncLimit=5  
# 數據存放文件夾，zookeeper運行過程中有兩個數據需要存儲，一個是快照數據（持久化數據）另一個是事務日志  
dataDir=/tmp/zookeeper  
## 客戶端訪問端口  
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  
# 啟動  
$ zkServer.sh start 

下面能看啟動成功

安裝 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 

•  啟動 Kafka 

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

大功告成！