Kafka - topic producer consumer 常用命令

Kafka常用命令

Topic相关

创建topic

bin/kafka-topics.sh --create --zookeeper  node02:2181 --replication-factor 1 --partitions 1 --topic test
1

kafka3.3.1新版本就会提示不兼容，需要改用新版本

Exception in thread "main" joptsimple.UnrecognizedOptionException: zookeeper is not a recognized option
        at joptsimple.OptionException.unrecognizedOption(OptionException.java:108)
        at joptsimple.OptionParser.handleLongOptionToken(OptionParser.java:510)
        at joptsimple.OptionParserState$2.handleArgument(OptionParserState.java:56)
        at joptsimple.OptionParser.parse(OptionParser.java:396)
        at kafka.admin.TopicCommand$TopicCommandOptions.(TopicCommand.scala:567)
        at kafka.admin.TopicCommand$.main(TopicCommand.scala:47)
        at kafka.admin.TopicCommand.main(TopicCommand.scala)

1
2
3
4
5
6
7
8
9
10

bin/kafka-topics.sh --create --bootstrap-server node02:9092 --replication-factor 1 --partitions 1 --topic test
1

创建多分区 副本topic

bin/kafka-topics.sh --create --bootstrap-server node02:9092 --replication-factor 1 --partitions 2 --topic test
1

查看topic

bin/kafka-topics.sh --list  --bootstrap-server node02:9092 
1

查看指定topic信息

bin/kafka-topics.sh --describe --bootstrap-server node02:9092  --topic test
1

可以进入kafka的数据文件存储目录查看test和test1主题的消息日志文件：默认log目录 /tmp/kafka-logs

消息日志文件主要存放在分区文件夹里的以log结尾的日志文件里，如下是test-1主题对应的分区0的消息日志：

leader节点负责给定partition的所有读写请求。
replicas 表示某个partition在哪几个broker上存在备份。不管这个几点是不是”leader“，甚至这个节点挂了，也会列出。
isr 是replicas的一个子集，它只列出当前还存活着的，并且已同步备份了该partition的节点。
1
2
3

删除主题

 bin/kafka-topics.sh --delete --topic test  --bootstrap-server node02:9092 
1

更新主题

bin/kafka-topics.sh -alter --partitions 1  --bootstrap-server node02:9092  --topic test
1

这个partition分区只能是增加分区，不可以降低，否则报错

Error while executing topic command : Topic currently has 2 partitions, which is higher than the requested 1.
[2022-12-03 20:45:18,531] ERROR org.apache.kafka.common.errors.InvalidPartitionsException: Topic currently has 2 partitions, which is higher than the requested 1.

1
2
3
4

producer发送消息

bin/kafka-console-producer.sh --broker-list node02:9092 --topic test 
1

consumer消费消息

bin/kafka-console-consumer.sh --bootstrap-server node02:9092 --topic test   
1

如果想要消费之前的消息可以通过–from-beginning参数指定，如下命令：

bin/kafka-console-consumer.sh --bootstrap-server node02:9092 --from-beginning --topic test 
1

消费多主题

bin/kafka-console-consumer.sh --bootstrap-server node02:9092 --whitelist "test|test-2"
1

同一个消费组下的消费者，只能有一个消费者消费到topic下的消息

设置消费组消息消息，单播消费

bin/kafka-console-consumer.sh --bootstrap-server node02:9092  --consumer-property group.id=testGroup --topic test 
1

多播消费，如消费订阅，不同消费组下的消费者可以消费同一topic下的同一条消息

bin/kafka-console-consumer.sh --bootstrap-server node02:9092 --consumer-property group.id=testGroup-2 --topic test 
1

列出消费组列表

bin/kafka-consumer-groups.sh --bootstrap-server node02:9092 --list
1

查看消费组，消费信息的消费偏移信息

bin/kafka-consumer-groups.sh --bootstrap-server node02:9092 --describe --group testGroup
1

current-offset：当前消费组的已消费偏移量
log-end-offset：主题对应分区消息的结束偏移量(HW)
lag：当前消费组未消费的消息数
1
2
3