kafka版本：kafka_2.11-0.9.0.1.tgz

kafka安裝節點：
cdhnode3 192.168.80.138
cdhnode4 192.168.80.139
cdhnode5 192.168.80.140

zookeeper所在節點
cdhnode1 192.168.80.136
cdhnode2 192.168.80.137
cdhnode3 192.168.80.148

• 1.解壓kafka壓縮包

tar -xzvf kafka_2.11-0.9.0.1.tgz

• 2.修改config中的service.properties配置檔案(註釋參考網上的資料)

#當前機器在叢集中的唯一標識，和zookeeper的myid性質一樣
broker.id=0

#當前kafka對外提供服務的埠預設是9092
port=19092

#這個引數預設是關閉的，在0.8.1有個bug，DNS解析問題，失敗率的問題。
host.name=192.168.80.138

#這個是borker進行網路處理的執行緒數
num.network.threads=3

#這個是borker進行I/O處理的執行緒數
num.io.threads=8

#訊息存放的目錄，這個目錄可以配置為“，”逗號分割的表示式，上面的num.io.threads要大於這個目錄的個數這個目錄
#如果配置多個目錄，新建立的topic他把訊息持久化的地方是，當前以逗號分割的目錄中，那個分割槽數最少就放那一個
 

log.dirs=/home/hadoop/app/kafka_2.11-0.10.1.1/logs

#傳送緩衝區buffer大小，資料不是一下子就傳送的，先回儲存到緩衝區了到達一定的大小後在傳送，能提高效能
socket.send.buffer.bytes=102400

#kafka接收緩衝區大小，當資料到達一定大小後在序列化到磁碟
socket.receive.buffer.bytes=102400

#這個引數是向kafka請求訊息或者向kafka傳送訊息的請請求的最大數，這個值不能超過java的堆疊大小
socket.request.max.bytes=104857600

#預設的分割槽數，一個topic預設1個分割槽數
 

num.partitions=1

#預設訊息的最大持久化時間，168小時，7天
log.retention.hours=168

#訊息儲存的最大值5M
message.max.byte=5242880

#kafka儲存訊息的副本數，如果一個副本失效了，另一個還可以繼續提供服務
default.replication.factor=2

#取訊息的最大直接數
replica.fetch.max.bytes=5242880

#這個引數是：因為kafka的訊息是以追加的形式落地到檔案，當超過這個值的時候，kafka會新起一個檔案
log.segment.bytes=1073741824

##每隔300000毫秒去檢查上面配置的log失效時間（log.retention.hours=168 ），到目錄檢視是否有過期的訊息如果有，刪除
log.retention.check.interval.ms=300000 

#是否啟用log壓縮，一般不用啟用，啟用的話可以提高效能
log.cleaner.enable=false 

#設定zookeeper的連線埠
zookeeper.connect=cdhnode1:2181,cdhnode2:2181,cdhnode3:2181 
message.max.byte=5242880
default.replication.factor=2
replica.fetch.max.bytes=5242880

• 3.分發kafka資料夾並啟動kafka叢集（三個都要啟動，提前先啟動zookeeper叢集）

分發kafka目錄
scp -r ../kafka_2.11-0.10.1.1/ hadoop@cdhnode5:/home/hadoop/app

啟動kafka（三臺機器都要啟動）
./bin/kafka-server-start.sh -daemon ./config/server.properties

• 4.檢視守護程序

如果kafka守護程序隔幾秒鐘又掛了，可以參考logs的server.log日誌目錄的錯誤資訊。

• 5.建立topic

./bin/kafka-topics.sh --create --zookeeper 192.168.80.136:2181,192.168.80.137:2181,192.168.80.138:2181 --replication-factor 2 --partitions 3 --topic lijietest

--create 建立

--zookeeper 連線的zookeeper

--replication-factor 備份數，類似於hadoop的冗餘策略

--partition 為topic分配多少個partition 

--topic 主題的名字

• 6.在主題lijietest上啟動一個生產者

./kafka-console-producer.sh --broker-list 192.168.80.138:19092,192.168.80.139:19092,192.168.80.140:19092 --topic lijietest

• 7.在主題lijietest啟動一個消費者

./kafka-console-consumer.sh --zookeeper 192.168.80.136:2181,192.168.80.137:2181,192.168.80.138:2181 --topic lijietest --from-beginning

在生產者端生產資料：

在消費者段消費資料：

到此為止，kafka叢集搭建完畢。

• 8.kafka的shell命令

1）.查詢所有的topic

./kafka-topics.sh --list --zookeeper 192.168.80.138:2181,192.168.80.139:2181,192.168.80.140:2181

2）.查詢topic的狀態

./kafka-topics.sh --describe --zookeeper 192.168.80.136:2181,192.168.80.137:2181,192.168.80.138:2181 --topic lijietest