一、 安裝JDK &配置JAVA_HOME
JDK安裝和JAVA_HOME配置，大家都應該很熟悉了，我們在此不再贅述。
二、 安裝Zookeeper
參照Zookeeper官網(wǎng)搭建一個ZK集群， 并啟動ZK集群。
三、 解壓Kafka安裝包
1. 修改配置文件config/server.properties
vi server.properties
broker.id=0 //集群中唯一id,依次增長：0、1、2、3、4
log.dirs=/root/kafkaData/ // 設置Kafka的消息數(shù)據(jù)存儲路徑
zookeeper.connect=node1:2181,node2:2181,node3:2181 //zookeeperServers列表，節(jié)點間以逗號分開
delete.topic.enable=true
listeners=PLAINTEXT:/node1:9092
log.dirs=/root/kafkaData/
num.partitions=3
# 往其他節(jié)點分發(fā)安裝包
for i in 2 3 ; do scp -r kafka_2.11-0.10.2.1/node$i:$PWD ;done
在其它幾臺節(jié)點上，分別修改此配置文件：broker.id依次增長。
2. 啟動Kafka
啟動kafka之前，必須要保證在每個節(jié)點上zookeeper都是啟動的。
在每個節(jié)點上啟動Kafka：
bin/kafka-server-start.sh [-daemon] config/server.properties
3. 集群測試
(1)進入kafka根目錄，創(chuàng)建test主題(topic)
bin/kafka-topics.sh --create --zookeeper node1:2181,node2:2181,node3:2181 --replication-factor 3 --partitions 1 --topic test
(2)顯示已創(chuàng)建的topic列表
bin/kafka-topics.sh --list --zookeeper node1:2181
(3)顯示topic的詳細信息
bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test
Topic:test PartitionCount:1 ReplicationFactor:3 Configs:
Topic: test Partition: 0 Leader: 1 Replicas: 1,2,0 Isr: 1,2,0
第一行是所有分區(qū)的一個整體描述，然后下面列出每個分期的信息，每個分區(qū)對應一行。因為本例中只有一個分區(qū)，所以下面只有一行。
leader：負責讀寫消息，zookeeper從所有節(jié)點中隨機選擇leader。
replicas：，列出所有的副本節(jié)點，即便節(jié)點現(xiàn)在不在服務中。
isr：列出正在服務中的節(jié)點。
在本例中，節(jié)點1是作為leader運行。
(4)使用客戶端命令去發(fā)送消息
bin/kafka-console-producer.sh --broker-list node1:9092,node2:9092,node3:9092 --topic test
(5)使用客戶端命令去消費消息
bin/kafka-console-consumer.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --from-beginning --topic test
可以通過參數(shù) consumer-property來指定消費者的group id：
kafka-console-consumer.sh --topic test --bootstrap-server node1:9092,node2:9092,node3:9092 --consumer-property group.id=xx
(6)測試集群容錯能力
Kill -9 pid[leader節(jié)點]
此時，另外一個節(jié)點被選做了leader, in-sync 副本列表中不再有節(jié)點1：
bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test
Topic:test PartitionCount:1 ReplicationFactor:3 Configs:
Topic: test Partition: 0 Leader: 2 Replicas: 1,2,0 Isr: 2,0
雖然最初負責讀寫消息的leader 不在線了，但之前的消息仍然可以消費：
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic test
(7)刪除主題
bin/kafka-topics.sh --delete --zookeeper node1:2181,node2:2181,node3:2181 --topic test
相關鏈接：
