Kafka常用命令 Topic相关

创建topic

创新互联公司专注于源汇企业网站建设,成都响应式网站建设公司,商城建设。源汇网站建设公司,为源汇等地区提供建站服务。全流程按需网站制作，专业设计，全程项目跟踪，创新互联公司专业和态度为您提供的服务

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

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)

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

创建多分区 副本topic

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

查看topic

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

查看指定topic信息

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

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

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

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

删除主题

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

更新主题

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

这个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.

参数	描述
–bootstrap-server	连接的 Kafka Broker 主机名称和端口号。
–topic	操作的 topic 名称。
–create	创建主题。
–delete	删除主题。
–alter	修改主题
–list	查看所有主题
–describe	查看主题详细描述
–partitions	设置分区数
–replication-factor	设置分区副本
–config	更新系统默认的配置。

producer发送消息

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

参数	描述
–bootstrap-server	连接的 Kafka Broker 主机名称和端口号
–topic	操作的topic

consumer消费消息

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

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

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

消费多主题

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

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

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

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

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

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

列出消费组列表

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

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

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

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

参数	描述
–bootstrap-server	连接的 Kafka Broker 主机名称和端口号
–topic	操作的 topic 名称
–from-beginning	从头开始消费
–group	指定消费者组名称。

你是否还在寻找稳定的海外服务器提供商？创新互联www.cdcxhl.cn海外机房具备T级流量清洗系统配攻击溯源，准确流量调度确保服务器高可用性，企业级服务器适合批量采购，新人活动首月15元起，快前往官网查看详情吧

本文名称：Kafka-topicproducerconsumer常用命令-创新互联
地址分享：http://kswjz.com/article/dhedgo.html