一、Kafka安裝與配置

步驟：

1、使用XFTP將Kafka安裝包kafka_2.12-2.8.1.tgz發送到master機器的主目錄。

2、解壓安裝包：

tar -zxvf ~/kafka_2.12-2.8.1.tgz

3、修改文件夾的名字，將其改為kafka，或者創建軟連接也可：

mv ~/kafka_2.12-2.8.1 ~/kafka

4、修改配置文件，Kafka包含生產者、消費者、ZooKeeper、Kafka集群節點等四個角色，因此需要修改對應的4個配置文件即可。Kafka的配置文件夾在解壓文件夾里面的config文件夾內，先執行以下命令進入到對應的目錄：

cd ~/kafka/config

（1）修改zookeeper.properties配置文件：

# 指定ZooKeeper的數據目錄
dataDir=/home/hadoop/zkdata
# 指定ZooKeeper的端口（這個應該不用改）
clientPort=2181

（2）修改consumer.properties配置文件：

# 配置Kafka集群地址
bootstrap.servers=master:9092,slave1:9092,slave2:9092

（3）修改producer.properties配置文件：

# 配置Kafka集群地址
bootstrap.servers=master:9092,slave1:9092,slave2:9092

（4）修改server.properties配置文件（找到對應的參數修改）：

# 指定ZooKeeper集群
zookeeper.connect=master:2181,slave1:2181,slave2:2181

上述內容修改完畢后保存。

5、將上面配置修改好后的Kafka安裝文件夾使用scp命令分發給slave1、slave2兩個節點：

scp -r ~/kafka hadoop@slave1:~
scp -r ~/kafka hadoop@slave2:~

6、分發完成以后，在每個節點上修改server編號：

cd ~/kafka/config
vim server.properties

找到broker.id項，master、slave1、slave2三臺機器分別標識為1、2、3：

broker.id=1

broker.id=2

broker.id=3

7、配置環境變量，三臺機器都要配置：

vim ~/.bashrc

?在文件末尾添加以下內容：

export KAFKA_HOME=/home/hadoop/kafka
export PATH=$KAFKA_HOME/bin:$PATH

保存文件，然后刷新環境變量或重新啟動命令行終端：

source ~/.bashrc

---

?

二、集群啟動測試

Kafka作為一個分布式應用程序，它依賴ZooKeeper提供協調服務，因此要啟動Kafka，必須先啟動ZooKeeper集群，再啟動Kafka集群。

步驟：

1、啟動好ZooKeeper以后，在三臺機器上執行以下命令啟動Kafka：

kafka-server-start.sh -daemon ~/kafka/config/server.properties

用jps命令在三臺機器上都能看到Kafka進程的話，說明Kafka集群服務啟動成功。

---

?

三、集群功能測試

集群啟動成功后，還需測試其功能是否能正確使用，Kafka自帶很多種Shell腳本供用戶使用，包括生產消息、消費消息、Topic管理等，下面利用Shell腳本來測試Kafka集群的功能。

步驟：

1、創建Topic：

kafka-topics.sh --zookeeper localhost:2181 --create --topic test --replication-factor 3 --partitions 3

2、查看Topic列表：

kafka-topics.sh --zookeeper localhost:2181 --list

能夠看到第1步創建的test則創建Topic成功。

3、啟動消費者：

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test

這樣就打開了一個終端消費者，消費test?Topic中的消息，但目前該Topic中還沒有消息，因此這個進程會沒有任何反應，這是正常的。

4、另外打開一個終端，啟動生產者向test?Topic發送消息：

kafka-console-producer.sh --broker-list localhost:9092 --topic test

在這個生產者終端上輸入消息，然后按回車發送，發送完之后，如果剛才那個消費者終端上，能夠查看到消費了對應的消息的話，說明Kafka集群可以正常對消息進行生產和消費了。

生產者slave1發送消息：

消費者master接受到消息：

?