余姚做网站公司企业网站建设方案策划

原创/朱季谦

接触kafka开发已经两年多，也看过关于kafka的一些书，但一直没有怎么对它做总结，借着最近正好在看《Apache Kafka实战》一书，同时自己又搭建了三台kafka服务器，正好可以做一些总结记录。

本文主要是记录如何在kafka集群服务器上创建topic，查看topic列表、分区、副本详情，删除topic，测试topic发送与消费，算是最基础的操作了，当然，不同版本其实指令是有所差异的，本文只针对kafka 2.x版本。虽然这些指令都很简单，但久不用了，很容易就会忘记，所谓好记性不如烂笔头，记录下来会更好。

主要写给一些刚接触kafka的小白，大神请自行绕路。

最后，推荐两本写kafka的书，我觉得挺不错的，一本是关于原理的《深入理解kafka:核心设计与实践原理》，一本就是我现在在看的，比较偏实际开发工作的《Apache Kafka实战》。这两本我觉得比较容易看懂，当然，我是先看的第一本，当时记录了一堆笔记，笔记目前放在我的GitHub上：GitHub - z924931408/Learning-Notes-document: 本仓库用于存储个人学习笔记。

我会一步一步记录，从简到原理。

1、创建一个名为test-topic的topic，该topic有3个分区，每个分区分配3个副本——

./kafka-topics.sh --zookeeper zk1:2181,zk2:2181,zk3:2181 --create --topic test-topic --partitions 3 --replication-factor 3

创建成功时，会提示：Created topic test-topic.

2、查看kafka集群已有topic列表——

./kafka-topics.sh --zookeeper zk1:2181,zk2:2181,zk3:2181 -list

查询情况会显示出所有的topic名——

3、查看topic的分区与备份详细情况——

./kafka-topics.sh --zookeeper zk1:2181,zk2:2181,zk3:2181 --describe --topic test-topic

例如，我在虚拟机上查看如下

PartitionCount：表示总共有3个分区；

ReplicationFactor：表示每个分区有3个副本备份；

该kafka集群分布部署在三台机器上，三个节点的id分别为0,1,2。截图当中，显示有0,1,2的字样，即分别代表了三台节点服务器，说明该test-topic三个分区各自均匀分配在三台机器上，分区的leader副本与三个follower副本同样均匀在三台机器上，这能起到防止数据丢失的作用。

最后，还有一个ISR，该参数全称，in-sync replica，它维护了一个集合，例如截图里的2,0,1，代表2,0,1副本保存的消息日志与leader 副本是保持一致的，只有保持一致的副本（包括所有副本），才会被维护在ISR集合里，当出现一定程度的不同步时，就会将该对应已经不一致的副本移出ISR集合，但是，这种移出并非永久的，一旦被移出的副本慢慢又恢复与leader一样时，那么，又会被加回isr集合当中。注意一点，只有在这个ISR里的副本服务器，才能在leader出现问题时有机会被选举为新的leader。

4、删除topic——

./kafka-topics.sh --zookeeper zk1:2181,zk2:2181,zk3:2181 --delete --topic test-topic

执行该语句后，会显示以下信息——

Topic test-topic is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.

翻译过来，就是说，主题test-topic标记为删除，但如果delete.topic.enable未设置为true，这将没有影响。

也就是说，如果delete.topic.enable未设置为true，即使执行了该语句，但也不会被删除，只会被标记一个删除状态而已，该delete.topic.enable参数在1.0.0版本后都是默认true，可以通过./kafka-topics.sh --zookeeper zk1:2181,zk2:2181,zk3:2181 -list指令查看，发现确实已经被删除了，当删除topic时，会将topic分区及备份所有数据都删除完，注意一点，这种删除是异步的。

5、验证集群消息发送与消费

kafka默认提供了两个脚本：kafka-console-producer.sh与kafka-console-consumer.sh。

可以直接使用这两个脚本验证集群是否能正常发送消息与消费消息。

开启一个生产者——

 ./kafka-console-producer.sh --broker-list kafka1:9092, kafka2:9092, kafka3:9092 --topic test-topic2

在其他两台机器上分别开启两个消费者——

 ./kafka-console-consumer.sh --bootstrap-server  kafka1:9092, kafka2:9092, kafka3:9092 --topic test-topic2 --from-beginning

大概效果如下——

生产者：

消费者：