技術標籤：springbootkafkaJavajavakafka

PS：歡迎轉載，但請註明出處，謝謝配合。

Kafka叢集搭建（基於kafka_2.11-1.0.0）

• 前言
• 一、基礎環境準備（CentOS，docker）
• 二、環境準備（JDK）
• 三、安裝 Kafka
• 四、使用 Kafka 叢集
• 五、參考

前言

Kafka叢集，本次部署3個節點。
本次3個節點都在同一臺機器上，實際生產環境要部署在多臺機器上，以保證高可用。
另外，zookeeper也只用了kafka內建的，只啟動了1個zookeeper節點，實際生產環境要在多個機器上部署zookeeper叢集，以保證zookeeper高可用。

一、基礎環境準備（CentOS，docker）

參考【Kafka學習筆記】一、Kafka單機搭建（基於kafka_2.11-1.0.0）

二、環境準備（JDK）

參考【Kafka學習筆記】一、Kafka單機搭建（基於kafka_2.11-1.0.0）

三、安裝 Kafka

1、獲取Kafka安裝包（離線方式）

將下載好的Kafka包，先放到共享目錄下，然後執行如下命令：    
cp /DockerContainerSharedFile/kafka_2.11-1.0.0.tgz /usr/local/    
cd /usr/local/    
補充說明：線上安裝，可參考官網命令：wget http://download.redis.io/releases/redis-6.0.6.tar.gz
 

2、解壓Redis原始碼包

tar -zxf kafka_2.11-1.0.0.tgz          
cd kafka_2.11-1.0.0

3、修改配置檔案
1）準備3個 kafka 服務配置檔案

cp config/server.properties config/server-9092.properties
cp config/server.properties config/server-9093.properties
cp config/server.properties config/server-9094.properties

2）修改內容如下

先建立儲存日誌資料的根目錄：
	mkdir -p /usr/local/kafka_2.11-1.0.0/data
--------------------------------------------
config/server-9092.properties:
--------------------------------------------
	broker.id=0                  【#當前機器在叢集中的唯一標識。注意：broker.id must be equal or greater than -1 and not greater than reserved.broker.max.id】
	listeners=PLAINTEXT://:9092   【#監聽埠】
	advertised.listeners=PLAINTEXT://127.0.0.1:9092    【監聽器釋出到Zookeeper供客戶端使用。若不配置，則當專案執行的伺服器和kafka執行的不是同一臺伺服器，會連線不上】
	log.dirs=/usr/local/kafka_2.11-1.0.0/data/kafka-logs-9092    【日誌資料的存放目錄】
--------------------------------------------
config/server-9093.properties:
--------------------------------------------
	broker.id=1
	listeners=PLAINTEXT://:9093
	advertised.listeners=PLAINTEXT://127.0.0.1:9093
	log.dirs=/usr/local/kafka_2.11-1.0.0/data/kafka-logs-9093
--------------------------------------------
config/server-9094.properties:
--------------------------------------------
	broker.id=2
	listeners=PLAINTEXT://:9094
	advertised.listeners=PLAINTEXT://127.0.0.1:9094
	log.dirs=/usr/local/kafka_2.11-1.0.0/data/kafka-logs-9094
 

4、啟動Zookeeper伺服器（Kafka內建）

nohup bin/zookeeper-server-start.sh config/zookeeper.properties >>zk.out &    【後臺啟動，退出，不終止】

啟動日誌：
[[email protected] kafka_2.11-1.0.0]# bin/zookeeper-server-start.sh config/zookeeper.properties
[2020-09-22 10:00:18,476] INFO Reading configuration from: config/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
......
[2020-09-22 10:00:18,551] INFO binding to port 0.0.0.0/0.0.0.0:2181 (org.apache.zookeeper.server.NIOServerCnxnFactory)

5、啟動Kafka伺服器（三臺）

nohup bin/kafka-server-start.sh config/server-9092.properties >>kafka-9092.out &    【後臺啟動，退出，不終止】
nohup bin/kafka-server-start.sh config/server-9093.properties >>kafka-9093.out &
nohup bin/kafka-server-start.sh config/server-9094.properties >>kafka-9094.out &

啟動日誌：
[[email protected] kafka_2.11-1.0.0]# bin/kafka-server-start.sh config/server-9092.properties
[2020-09-22 10:08:16,763] INFO KafkaConfig values:
......
[2020-09-22 10:08:17,716] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)

四、使用 Kafka 叢集

1、建立一個副本為3的新topic

# bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic testTopic

2、通過"describe topic"命令檢視叢集情況

# bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic testTopic 【後面的 --topic test 不加也行，就是輸出的內容會比較多】

以下是對輸出資訊的解釋。第一行給出了所有分割槽的摘要，下面的每行都給出了一個分割槽的資訊。因為我們只有一個分割槽，所以只有一行。

“leader”是負責給定分割槽所有讀寫操作的節點。每個節點都是隨機選擇的部分分割槽的領導者。
“replicas”是複製分割槽日誌的節點列表，不管這些節點是leader還是僅僅活著。
“isr”是一組“同步”replicas，是replicas列表的子集，它活著並被指到leader。

3、使用producer（生產者）傳送訊息

# bin/kafka-console-producer.sh --broker-list localhost:9092 --topic testTopic
>message A
>message B

4、使用consumer（消費者）獲取訊息（3臺都能檢視到）

# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testTopic --from-beginning
message A
message B

# bin/kafka-console-consumer.sh --bootstrap-server localhost:9093 --topic testTopic --from-beginning
message A
message B

# bin/kafka-console-consumer.sh --bootstrap-server localhost:9094 --topic testTopic --from-beginning
message A
message B

5、測試容錯性

1）根據上面的 kafka-topics.sh --describe 命令結果，得知 節點1 是該主題中唯一分割槽的領導者。通過程序ID，殺掉該程序

[[email protected] kafka_2.11-1.0.0]# ps ef | grep server-9092.properties
792 pts/1    Sl     2:10  \_ /usr/java/jdk1.8.0_261/bin/java -Xmx1G -Xms1G -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHea......
[[email protected] kafka_2.11-1.0.0]# kill -9 792

2）通過 kafka-topics.sh --describe 命令，重新檢視叢集情況（領導權已經切換到 Broker 2 ，且 Broker 1 也不再同步副本集中了 ）

3）重啟啟動 Broker 1 ，則會重新加入同步副本集

注意：
a）Leader Broker 停止後，之前的訊息仍可以消費（通過訪問其他 Broker ）
b）某個 Broker 停止後，新寫入的訊息，在該 Broker 重新啟動加入同步副本集後，會同步這些訊息。訪問該 Broker 可以獲取所有訊息。

五、參考

1、Kafka官網