Kafka叢集是把狀態儲存在Zookeeper中的，首先要搭建Zookeeper叢集。由於我們之前的分散式系統中已經安裝zookeeper服務，這裡不進行zookeeper安裝教程以及應用教程圖解 [https://blog.csdn.net/qq_34898847/article/details/83349679 ] 。需要注意的是：Zookeeper叢集的工作是超過半數才能對外提供服務，3臺中超過兩臺超過半數，允許1臺掛掉 ，是否可以用偶數，其實沒必要。如果有四臺那麼掛掉一臺還剩下三臺伺服器，如果在掛掉一個就不行了，這裡記住是超過半數。

1.軟體環境

linux一臺
已經搭建好的zookeeper叢集
[

2. 建立目錄並安裝

我的是普通使用者:
/home/xzb
cd ~/software/
cp kafka_2.11-2.0.0.tgz ./

• 解壓 tar -zxvf kafka_2.11-2.0.0.tgz
  mv kafka_2.11-2.0.0.tgz kafka2.0.0
• 準備日誌存放目錄
  cd ~/data/kafka
  每個kafka服務建立一個日誌存放目錄
  mkdir kafka-logs-{0,1,2}

3.建立server配置檔案副本

cp config/server.properties config/server-1.properties
cp config/server.properties config/server-2.properties
分別編輯一下幾個地方值得修改:
vim server.properties

broker.id=0
port=9092
log.dir=/home/xzb/data/kafka/kafka-logs-0
host.name=cent
zookeeper.connect=cent:2181

vim server-1.properties

broker.id=1
port=9093
log.dir=/home/xzb/data/kafka/kafka-logs-1
host.name=cent
zookeeper.connect=cent:2181

vim server-2.properties

broker.id=2
port=9094
log.dir=/home/xzb/data/kafka/kafka-logs-2
host.name=cent
zookeeper.connect=cent:2181
 

server.properties配置檔案引數資訊

#########################引數解釋##############################

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

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

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

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

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

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

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

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

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

num.partitions=1 #預設的分割槽數，一個topic預設1個分割槽數

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

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

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

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

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

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

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

zookeeper.connect=192.168.25.139:2181,192.168.25.139:2182,192.168.25.139:2183 #設定zookeeper的連線埠

#########################引數解釋##############################

 

#########################引數修改############################## 

broker.id=0  #每臺伺服器的broker.id都不能相同

host.name=192.168.7.100 #hostname

#在log.retention.hours=169下面新增下面三項

message.max.byte=5242880

default.replication.factor=2

replica.fetch.max.bytes=5242880

#設定zookeeper的連線埠

zookeeper.connect=192.168.7.100:12181,192.168.7.101:12181,192.168.7.107:12181

#########################引數修改##############################

4. 開放埠號

root 使用者
vi /etc/sysconfigs/iptables
新增:kafka服務
-A INPUT -m state --state NEW -m tcp -p tcp --dport 9092 -j ACCEPT
-A INPUT -m state --state NEW -m tcp -p tcp --dport 9093 -j ACCEPT
-A INPUT -m state --state NEW -m tcp -p tcp --dport 9094 -j ACCEPT

5. 啟動

• 1.啟動zookeeper叢集

• 2.啟動kafka叢集

     bin/kafka-server-start.sh config/server.properties & 
     bin/kafka-server-start.sh config/server-1.properties & 
     bin/kafka-server-start.sh config/server-2.properties &
  

• 3檢查kafka是否啟動

     jps
  

6. 測試

• 建立topic來驗證是否建立成功
  bin/kafka-topics.sh --create --zookeeper 192.168.25.139:2181, 192.168.25.139:2182,192.168.25.139:2183 --replication-factor 2 --partitions 1 --topic test

    --replication-factor 2   #複製兩份
    --partitions 1 #建立1個分割槽
    --topic #主題為test
  

• 在一臺伺服器上建立一個釋出者’’’#建立一個broker，釋出者

    bin/kafka-console-producer.sh --broker-list 192.168.25.139:2181, 
    192.168.25.139:2182,192.168.25.139:2183 --topic test
  

• ‘’‘在一臺伺服器上建立一個訂閱者’’’

    bin/kafka-console-consumer.sh --zookeeper 192.168.25.139:2181, 
    192.168.25.139:2182,192.168.25.139:2183 
     --topic test  --from-beginning
  

• 檢視topic

    bin/kafka-topics.sh --describe --zookeeper 192.168.25.139:2181, 
    192.168.25.139:2182,192.168.25.139:2183 --topic test
  

• 殺死其中一個broker之後,再次檢視主題資訊,出現新的leader則成功

7. 日誌說明

kafka的日誌是儲存在/usr/fuyou/kafka/kafka_2.10-0.10.1.0/logs目錄下的

 #kafka的執行日誌
 server.log 
 #kafka他是用zookeeper來儲存狀態，所以他可能會進行切換，切換的日誌就儲存在這裡
 state-change.log  

 #kafka選擇一個節點作為“controller”,當發現有節點down掉的時候它負責
 在游泳分割槽的所有節點中選擇新的leader,
 這使得Kafka可以批量的高效的管理所有分割槽節點的主從關係。
 如果controller down掉了，活著的節點中的一個會備切換為新的controller.
 controller.log