kafka基本命令查看博客《kafka2.5.0基本命令》
本博文所使用kafka版本2.5.0，操作系统centos8.
1）创建主题
创建my-topic主题，该主题有 1 个副本，8个分区：
$ bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 8 --topic my-topic Created topic my-topic.
注意这里的 --replication-factor 副本个数不能大于broker的个数
2）列出主题
$ bin/kafka-topics.sh --list --bootstrap-server localhost:9092
my-topic
test
3) 更改主题分区个数：
注意：分区个数只能改多，不能改少，这里从8个分区改为16个：
$ bin/kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic my-topic --partitions 16 artitions 16
验证分区扩建，cd 到kafka默认日志数据目录下： /tmp/kafka-logs，可以查看到16个分区文件夹：
drwxrwxr-x. 2 joyce joyce 141 Jun 21 05:17 my-topic-0 drwxrwxr-x. 2 joyce joyce 141 Jun 21 05:17 my-topic-1 drwxrwxr-x. 2 joyce joyce 141 Jun 21 05:17 my-topic-2 drwxrwxr-x. 2 joyce joyce 141 Jun 21 05:17 my-topic-3
.......
每个分区文件夹的核心文件是.log文件：
[joyce@192 kafka-logs]$ cd my-topic-0 [joyce@192 my-topic-0]$ ll total 4 -rw-rw-r--. 1 joyce joyce 10485760 Jun 21 05:17 00000000000000000000.index -rw-rw-r--. 1 joyce joyce 0 Jun 21 05:17 00000000000000000000.log // log日志过期时间可以在server.properties里配置：log.retention.hours=168
// 注意：在分区文件夹目录下，有个00000000000000000000.log文件，如果size大到一定程度，会新建一个10这样二进制命名的.log文件，例如10101010.log，00000000000000000000.log则停止更新，当旧文件的最后更新时间 + 日志保存时间 = 删除时间，00000000000000000000.log日志就会被删除。以此类推。
-rw-rw-r--. 1 joyce joyce 10485756 Jun 21 05:17 00000000000000000000.timeindex -rw-rw-r--. 1 joyce joyce 8 Jun 21 05:17 leader-epoch-checkpoint
查看分区当前情况：
[joyce@192 kafka_2.12-2.5.0]$ bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe Topic: kafka-boot PartitionCount: 1 ReplicationFactor: 1 Configs: segment.bytes=1073741824 Topic: kafka-boot Partition: 0 Leader: 0 Replicas: 0 Isr: 0 Topic: my-topic PartitionCount: 16 ReplicationFactor: 1 Configs: segment.bytes=1073741824 Topic: my-topic Partition: 0 Leader: 0 Replicas: 0 Isr: 0 Topic: my-topic Partition: 1 Leader: 0 Replicas: 0 Isr: 0 Topic: my-topic Partition: 2 Leader: 0 Replicas: 0 Isr: 0 Topic: my-topic Partition: 3 Leader: 0 Replicas: 0 Isr: 0
......
4）生产者
需连接多个broker的话，中间用逗号分隔：
$ bin/kafka-console-producer.sh --bootstrap-server localhost:9092,192.168.2.60:9092 --topic my-topic
5）消费者
消费者消费消息：
$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --from-beginning
如果加了--from-beginning，则从头开始消费消息
6）删除Topic
注意：尽量不要删除Topic，可能会弄坏kafka
kafka2.5.0版本里，server.properties已经没有 delete.topic.enable=true 这个配置，所以忽略此项。
步骤1：停止该Topic所有的生产者和消费者。
步骤2：进入kafka的bin目录下，用命令删除 Topic test1。此时可以在zookeeper里看到 admin --> delete-topic下，test1 被标记为删除，但是没有进行物理删除。
bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic test1
步骤3：删除kafka存储Topic目录，该目录默认配置在server.properties里的log.dirs=/tmp/log-kafka，进入该目录可以看到Topic的相关分区：test1-0、test1-1
步骤4：删除zookeeper里的topic信息：
bin/zkCli.sh -server 【zookeeper server:port】
登录到zk shell，然后找到topic所在目录： ls /brokers/topics，找到要删除的topic，然后执行命令：
rmr /brokers/topics/【topic name】
rmr /admin/delete_topics/ 【topic name】
bin/zkCli.sh -server localhost:2181 [zk: localhost:2181(CONNECTED) 0] ls /brokers/topics [__consumer_offsets, abc, my-topic, test1] [zk: localhost:2181(CONNECTED) 1] deleteall /brokers/topics/test1 // deleteall 命令可以删除该文件夹，即使目录不为空 [zk: localhost:2181(CONNECTED) 1] deleteall /admin/delete_topics/test1
步骤5：
重启zookeeper:
./zkServer.sh start
重启kafka:
bin/kafka-server-start.sh config/server.properties
end.