创建TOPIC

cd /usr/local/bin/kafka/kafka_2.12-0.11.0.1/bin/
./kafka-topics.sh  --create --zookeeper centos_1:2181  --replication-factor 1 --partitions 1 --topic wangcheng_test

查看TOPIC是否创建成功

./kafka-topics.sh --list --zookeeper centos_1:2181 wangcheng_test

生产者发送消息

./kafka-console-producer.sh --broker-list centos_1:9092 --topic wangcheng_test

消费者消费消息（可以在集群任何一台机器上消费）

./kafka-console-consumer.sh --zookeeper localhost:2181 --topic wangcheng_test --from-beginning

到此为止说明集群正确

创建多备份的topic

1、创建3个备份的topic
./kafka-topics.sh  --create --zookeeper centos_1:2181  --replication-factor 3 --partitions 1 --topic topic_1
2、查看topic_1的状态
./kafka-topics.sh --describe --zookeeper centos_2:2181 --topic topic_1
Topic:topic_1   PartitionCount:1        ReplicationFactor:3     Configs:
Topic: topic_1  Partition: 0    Leader: 2       Replicas: 2,1,3 Isr: 2,1,3

可以看到当前的Partition数量为1，ReplicationFactor备份数为3，leader是2
向centos_2中的kafka发信息

./kafka-console-producer.sh --broker-list centos_2:9092 --topic topic_1
topic_1_test

我们将centos_2上的kafka进程关闭,在centos_2上开启一个consumer

./kafka-console-consumer.sh --zookeeper localhost:2181 --topic topic_1 --from-beginning
Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper].
topic_1_test

再查看下topic_1的status

 ./kafka-topics.sh --describe --zookeeper centos_2:2181 --topic topic_1
Topic:topic_1   PartitionCount:1        ReplicationFactor:3     Configs:
        Topic: topic_1  Partition: 0    Leader: 1       Replicas: 2,1,3 Isr: 1,3

leader变成了1，ISR变成了1,3
重启2，再查看状态

./kafka-topics.sh --describe --zookeeper centos_2:2181 --topic topic_1
Topic:topic_1   PartitionCount:1        ReplicationFactor:3     Configs:
 Topic: topic_1  Partition: 0    Leader: 1       Replicas: 2,1,3 Isr: 1,3,2

Isr变回了1,3,2

再试下先关闭centos_1上的进程，发送再重启看下现象

[root@centos_1 bin]# jps -l
11525 kafka.Kafka
14341 sun.tools.jps.Jps
7532 org.apache.zookeeper.server.quorum.QuorumPeerMain
[root@centos_1 bin]# kill 11525
[root@centos_1 bin]# jps -l
14357 sun.tools.jps.Jps
7532 org.apache.zookeeper.server.quorum.QuorumPeerMain
[root@centos_1 bin]# ./kafka-topics.sh --describe --zookeeper centos_2:2181 --topic topic_1
Topic:topic_1   PartitionCount:1        ReplicationFactor:3     Configs:
 Topic: topic_1  Partition: 0    Leader: 2       Replicas: 2,1,3 Isr: 3,2
[root@centos_1 bin]# ./kafka-console-producer.sh --broker-list centos_2:9092 --topic topic_1
>2345678
[root@centos_2 bin]# ./kafka-console-consumer.sh --zookeeper localhost:2181 --topic topic_1 --from-beginning
Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper].
topic_1_test
2345678
[root@centos_1 bin]# ./kafka-server-start.sh ../config/server.properties &
[root@centos_1 bin]# ./kafka-topics.sh --describe --zookeeper centos_2:2181 --topic topic_1
Topic:topic_1   PartitionCount:1        ReplicationFactor:3     Configs:
        Topic: topic_1  Partition: 0    Leader: 2       Replicas: 2,1,3 Isr: 3,2,1

根据以上现象说明，broker重启会重新拉取副本信息
参考资料：
kafka副本同步机制
0.10.0后版本的参数修改
kafka文件存储机制