一.前期準備

1.1 Win7官網下載kafka包

本文使用版本kafka_2.10-0.10.1.0.tgz

1.2 配置jdk、scala、zookeeper

jdk,scala,kafka版本要對應

1.3 centos7叢集伺服器

主機名    系統          IP地址

master    centos7    192.168.32.128

slave01   centos7    192.168.32.131

slave02   centos7     192.168.32.132

二.kafka叢集搭建

以下操作只針對master主機伺服器，其他主機伺服器類似。

2.1 上傳kafka包至 /opt/software目錄

2.2 解壓和拷貝kafka至 /usr/local/kafka

cd /opt/software
tar -zxvf kafka_2.10-0.10.1.0.tgz
cp -r kafka_2.10-0.10.1.0 /usr/local/kafka

三.kafka叢集配置

3.1 server.properties檔案配置

進入kafka中的config目錄

vi server.properties

#switch to enable topic deletion or not, default value is false
delete.topic.enable=true
#本機對應的ip地址
listeners=PLAINTEXT://192.168.32.128:9092
log.dirs=/tmp/kafka-logs
zookeeper.connect=192.168.32.128:2181,192.168.32.132:2181,192.168.32.131:2181
 

注意：相關配置項不要重複。

3.2 新增/tmp/kafka-logs

mkdir -p /tmp/kafka-logs

至此 master主機伺服器kafka已搭建完成。

3.3 其他伺服器kafka搭建

搭建192.168.32.131/slave01和192.168.32.132/slave02伺服器的kafka

拷貝master伺服器的kafka至slave01和slave02的 /usr/local/目錄

master伺服器：

cd /usr/local
scp -r kafka [email protected]:/usr/local/

slave01伺服器：

類似3.1配置server.properties

注意：

broker.id=1,要與master中broker.id區別

listeners=PLAINTEXT://192.168.32.131:9092,需要改成本機對應ip

slave02伺服器配置類似。

四.kafka叢集測試

4.1 測試命令

Step 1: Start the server

後臺方式啟動，推薦第一次配置的新手不要加入-daemon引數，看看控制檯輸出的是否有success.

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

Step 2: Create a topic(replication-factor一定要大於1，否則kafka只有一份資料，leader一旦崩潰程式就沒有輸入源了，分割槽數目視輸入源而定)

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

Step 3: Describe a topic

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic topicTest

step 4: list the topic

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

step 5: send some message

bin/kafka-console-producer.sh --broker-list localhost:2181 --topic topicTest

step 6: start a consumer

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic topicTest --from-beginning

step 7: delete a topic

要事先在 serve.properties 配置 delete.topic.enable=true

bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic topicTest

# 如果仍然只是僅僅被標記了刪除(zk中並沒有被刪除)，那麼啟動zkCli.sh,輸入如下指令

rmr /brokers/topics/topicTest

4.2 叢集測試

step1:啟動zookeeper叢集

step2:啟動kafka叢集

三臺伺服器分別執行啟動命令

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

jps檢視程序

注意：kafka是kafka程序,QuorumPeerMain是zookeeper程序

step3. 建立主題和檢視主題

bin/kafka-topics.sh --create --zookeeper 192.168.32.128:2181,192.168.32.131:2181,192.168.32.132:2181 --replication-factor 3 --partitions 3 --topic topicTest

bin/kafka-topics.sh --list --zookeeper 192.168.32.128:2181,192.168.32.131:2181,192.168.32.132:2181

step4. 啟動訊息生產者和訊息消費者

master伺服器

bin/kafka-console-producer.sh --broker-list 192.168.32.128:9092,192.168.32.131:9092,192.168.32.132:9092 --topic topicTest

slave01或slave02伺服器

bin/kafka-console-consumer.sh --zookeeper 192.168.32.128:2181,192.168.32.131:2181,192.168.32.132:2181 --topic topicTest --from-beginning

master伺服器輸入資訊，slave01或slave02會顯示master輸的的資訊