(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

第一行是所有分區的一個整體描述，然后下面列出每個分期的信息，每個分區對應一行。因為本例中只有一個分區，所以下面只有一行。

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

replicas：，列出所有的副本節點，即便節點現在不在服務中。

isr：列出正在服務中的節點。

在本例中，節點1是作為leader運行。

(4)使用客戶端命令去發送消息

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

可以通過參數 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節點]

此時，另外一個節點被選做了leader, in-sync 副本列表中不再有節點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