(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ū)對應(yīng)一行。因為本例中只有一個分區(qū)，所以下面只有一行。

leader：負責(zé)讀寫消息，zookeeper從所有節(jié)點中隨機選擇leader。

replicas：，列出所有的副本節(jié)點，即便節(jié)點現(xiàn)在不在服務(wù)中。

isr：列出正在服務(wù)中的節(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

雖然最初負責(zé)讀寫消息的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