Kafka集羣搭建及必知必會

Kafka集羣部署並啓動

在本文中將從演示如何搭建一個Kafka集羣開始，而後簡要介紹一下關於Kafka集羣的一些基礎知識點。但本文僅針對集羣作介紹，對於Kafka的基本概念不作過多說明，這裏假設讀者擁有必定的Kafka基礎知識。

首先，咱們須要瞭解Kafka集羣的一些機制：

• Kafka是自然支持集羣的，哪怕是一個節點實際上也是集羣模式
• Kafka集羣依賴於Zookeeper進行協調，而且在早期的Kafka版本中不少數據都是存放在Zookeeper的
• Kafka節點只要註冊到同一個Zookeeper上就表明它們是同一個集羣的
• Kafka經過brokerId來區分集羣中的不一樣節點

Kafka的集羣拓撲圖以下：

Kafka集羣中的幾個角色：

• Broker：通常指Kafka的部署節點
• Leader：用於處理消息的接收和消費等請求，也就是說producer是將消息push到leader，而consumer也是從leader上去poll消息
• Follower：主要用於備份消息數據，一個leader會有多個follower

在本例中，爲了更貼近實際的部署狀況，使用了四臺虛擬機做演示：

機器IP	做用	角色	brokerId
192.168.99.1	部署Kafka節點	broker server	0
192.168.99.2	部署Kafka節點	broker server	1
192.168.99.3	部署Kafka節點	broker server	2
192.168.99.4	部署Zookeeper節點	集羣協調者

Zookeeper安裝

Kafka是基於Zookeeper來實現分佈式協調的，因此在搭建Kafka節點以前須要先搭建好Zookeeper節點。而Zookeeper和Kafka都依賴於JDK，我這裏已經事先安裝好了JDK：

[root@192.168.99.4 ~]# java --version
java 11.0.5 2019-10-15 LTS
Java(TM) SE Runtime Environment 18.9 (build 11.0.5+10-LTS)
Java HotSpot(TM) 64-Bit Server VM 18.9 (build 11.0.5+10-LTS, mixed mode)
[root@txy-server2 ~]#

準備好JDK環境後，到Zookeeper的官網下載地址，複製下載連接：

• https://zookeeper.apache.org/releases.html#download

而後到Linux中使用wget命令進行下載，以下：

[root@192.168.99.4 ~]# cd /usr/local/src
[root@192.168.99.4 /usr/local/src]# wget https://mirror.bit.edu.cn/apache/zookeeper/zookeeper-3.6.1/apache-zookeeper-3.6.1-bin.tar.gz

解壓下載好的壓縮包，並將解壓後的目錄移動和重命名：

[root@192.168.99.4 /usr/local/src]# tar -zxvf apache-zookeeper-3.6.1-bin.tar.gz
[root@192.168.99.4 /usr/local/src]# mv apache-zookeeper-3.6.1-bin ../zookeeper

進入到Zookeeper的配置文件目錄，將zoo_sample.cfg這個示例配置文件拷貝一份並命名爲zoo.cfg，這是Zookeeper默認的配置文件名稱：

[root@192.168.99.4 /usr/local/src]# cd ../zookeeper/conf/
[root@192.168.99.4 /usr/local/zookeeper/conf]# ls
configuration.xsl  log4j.properties  zoo_sample.cfg
[root@192.168.99.4 /usr/local/zookeeper/conf]# cp zoo_sample.cfg zoo.cfg

修改一下配置文件中的dataDir配置項，指定一個磁盤空間較大的目錄：

[root@192.168.99.4 /usr/local/zookeeper/conf]# vim zoo.cfg
# 指定Zookeeper的數據存儲目錄，類比於MySQL的dataDir
dataDir=/data/zookeeper
[root@192.168.99.4 /usr/local/zookeeper/conf]# mkdir -p /data/zookeeper

若是隻是學習使用的話，這一步其實能夠忽略，採用默認配置便可
接下來就能夠進入bin目錄，使用啓動腳原本啓動Zookeeper了，以下示例：

[root@192.168.99.4 /usr/local/zookeeper/conf]# cd ../bin/
[root@192.168.99.4 /usr/local/zookeeper/bin]# ./zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED
[root@192.168.99.4 /usr/local/zookeeper/bin]#

啓動完成後，能夠經過查看是否正常監聽端口號來判斷是否啓動成功，以下則是啓動成功了：

[root@192.168.99.4 ~]# netstat -lntp |grep 2181
tcp6       0      0 :::2181       :::*         LISTEN      7825/java
[root@192.168.99.4 ~]#

若是你的機器開啓了防火牆的話，則須要開放Zookeeper的端口，不然其餘節點沒法註冊上來：

[root@192.168.99.4 ~]# firewall-cmd --zone=public --add-port=2181/tcp --permanent
[root@192.168.99.4 ~]# firwall-cmd --reload

Kafka安裝

安裝完Zookeeper後，接下來就能夠安裝Kafka了，一樣的套路首先去Kafka的官網下載地址，複製下載連接：

• https://zookeeper.apache.org/releases.html#download

而後到Linux中使用wget命令進行下載，以下：

[root@192.168.99.1 ~]# cd /usr/local/src
[root@192.168.99.1 /usr/local/src]# wget https://mirror.bit.edu.cn/apache/kafka/2.5.0/kafka_2.13-2.5.0.tgz

解壓下載好的壓縮包，並將解壓後的目錄移動和重命名：

[root@192.168.99.1 /usr/local/src]# tar -xvf kafka_2.13-2.5.0.tgz
[root@192.168.99.1 /usr/local/src]# mv kafka_2.13-2.5.0 ../kafka

進入Kafka的配置文件目錄，修改配置文件：

[root@192.168.99.1 /usr/local/src]# cd ../kafka/config/
[root@192.168.99.1 /usr/local/kafka/config]# vim server.properties
# 指定該節點的brokerId，同一集羣中的brokerId須要惟一
broker.id=0
# 指定監聽的地址及端口號，該配置項是指定內網ip
listeners=PLAINTEXT://192.168.99.1:9092
# 若是須要開放外網訪問，則在該配置項指定外網ip
advertised.listeners=PLAINTEXT://192.168.99.1:9092
# 指定kafka日誌文件的存儲目錄
log.dirs=/usr/local/kafka/kafka-logs
# 指定zookeeper的鏈接地址，如有多個地址則用逗號分隔
zookeeper.connect=192.168.99.4:2181
[root@192.168.99.1 /usr/local/kafka/config]# mkdir /usr/local/kafka/kafka-logs

在完成配置文件的修改後，爲了方便使用Kafka的命令腳本，咱們能夠將Kafka的bin目錄配置到環境變量中：

[root@192.168.99.1 ~]# vim /etc/profile
export KAFKA_HOME=/usr/local/kafka
export PATH=$PATH:$KAFKA_HOME/bin
[root@192.168.99.1 ~]# source /etc/profile  # 讓配置生效

這樣就可使用以下命令啓動Kafka了：

[root@192.168.99.1 ~]# kafka-server-start.sh /usr/local/kafka/config/server.properties &

執行以上命令後，啓動日誌會輸出到控制檯，能夠經過日誌判斷是否啓動成功，也能夠經過查看是否監聽了9092端口來判斷是否啓動成功：

[root@192.168.99.1 ~]# netstat -lntp |grep 9092
tcp6    0     0 192.168.99.1:9092     :::*      LISTEN     31943/java
[root@192.168.99.1 ~]#

一樣的，開啓了防火牆的話，還須要開放相應的端口號：

[root@192.168.99.1 ~]# firewall-cmd --zone=public --add-port=9092/tcp --permanent
[root@192.168.99.1 ~]# firwall-cmd --reload

到此爲止，咱們就完成了第一個Kafka節點的安裝，另外兩個節點的安裝步驟也是同樣的，只須要修改一下配置文件中的brokerId和監聽的ip就行了。因此我這裏直接將該節點中的Kafka目錄拷貝到另外兩臺機器上：

[root@192.168.99.1 ~]# rsync -av /usr/local/kafka 192.168.99.2:/usr/local/kafka
[root@192.168.99.1 ~]# rsync -av /usr/local/kafka 192.168.99.3:/usr/local/kafka

而後修改一下這兩個節點的brokerId和監聽的ip：

[root@192.168.99.2 /usr/local/kafka/config]# vim server.properties
# 修改brokerId
broker.id=1
# 指定監聽的地址及端口號，該配置項是指定內網ip
listeners=PLAINTEXT://192.168.99.2:9092
# 若是須要開放外網訪問，則在該配置項指定外網ip
advertised.listeners=PLAINTEXT://192.168.99.2:9092
[root@192.168.99.2 /usr/local/kafka/config]# 
[root@192.168.99.3 /usr/local/kafka/config]# vim server.properties
# 修改brokerId
broker.id=2
# 指定監聽的地址及端口號，該配置項是指定內網ip
listeners=PLAINTEXT://192.168.99.3:9092
# 若是須要開放外網訪問，則在該配置項指定外網ip
advertised.listeners=PLAINTEXT://192.168.99.3:9092
[root@192.168.99.3 /usr/local/kafka/config]#

配置修改完成後，按以前所介紹的步驟啓動這兩個節點。啓動成功後進入Zookeeper中，在/brokers/ids下有相應的brokerId數據表明集羣搭建成功：

[root@192.168.99.4 ~]# /usr/local/zookeeper/bin/zkCli.sh
[zk: localhost:2181(CONNECTED) 4] ls /brokers/ids
[0, 1, 2]
[zk: localhost:2181(CONNECTED) 5]

Kafka副本集

關於Kafka的副本集：

• Kafka副本集是指將日誌複製多份，咱們知道Kafka的數據是存儲在日誌文件中的，這就至關於數據的備份、冗餘
• Kafka能夠經過配置設置默認的副本集數量
• Kafka能夠爲每一個Topic設置副本集，因此副本集是相對於Topic來講的
• 一個Topic的副本集能夠分佈在多個Broker中，當一個Broker掛掉了，其餘的Broker上還有數據，這就提升了數據的可靠性，這也是副本集的主要做用。

咱們都知道在Kafka中的Topic只是個邏輯概念，實際存儲數據的是Partition，因此真正被複制的也是Partition。以下圖：

關於副本因子：

• 副本因子其實決定了一個Partition的副本數量，例如副本因子爲1，則表明將Topic中的全部Partition按照Broker的數量複製一份，並分佈到各個Broker上

副本分配算法以下：

• 將全部N Broker和待分配的i個Partition排序
• 將第i個Partition分配到第(i mod n)個Broker上
• 將第i個Partition的第j個副本分配到第((i + j) mod n)個Broker上

Kafka節點故障緣由及處理方式

Kafka節點（Broker）故障的兩種狀況：

• Kafka節點與Zookeeper心跳未保持視爲節點故障
• 當follower的消息落後於leader太多也會視爲節點故障

Kafka對節點故障的處理方式：

• Kafka會對故障節點進行移除，因此基本不會由於節點故障而丟失數據
• Kafka的語義擔保也很大程度上避免了數據丟失
• Kafka會對消息進行集羣內平衡，減小消息在某些節點熱度太高

Kafka Leader選舉機制簡介

Kafka集羣之Leader選舉：

• 若是有接觸過其餘一些分佈式組件就會了解到大部分組件都是經過投票選舉來在衆多節點中選舉出一個leader，但在Kafka中沒有采用投票選舉來選舉leader
• Kafka會動態維護一組Leader數據的副本（ISR）
• Kafka會在ISR中選擇一個速度比較快的設爲leader

「巧婦難爲無米之炊」：Kafka有一種無奈的狀況，就是ISR中副本所有宕機。對於這種狀況，Kafka默認會進行unclean leader選舉。Kafka提供了兩種不一樣的方式進行處理：

1. 等待ISR中任一Replica恢復，並選它爲Leader

   • 等待時間較長，會下降可用性，或ISR中的全部Replica都沒法恢復或者數據丟失，則該Partition將永不可用
2. 選擇第一個恢復的Replica爲新的Leader，不管它是否在ISR中
   • 並未包含全部已被以前Leader Commit過的消息，所以會形成數據丟失，但可用性較高

Leader選舉配置建議：

• 禁用unclean leader選舉
• 手動設置最小ISR

關於ISR更詳細的內容能夠參考：

• https://www.jianshu.com/p/ff296d51385a
• http://www.javashuo.com/article/p-daptuxov-bg.html

原文連接：https://www.jianshu.com/p/cc0b90636715 做者：端碗吹水