Kafka常用命令 Topic相關(guān)

創(chuàng)建topic

創(chuàng)新互聯(lián)公司專注于源匯企業(yè)網(wǎng)站建設(shè),成都響應(yīng)式網(wǎng)站建設(shè)公司,商城建設(shè)。源匯網(wǎng)站建設(shè)公司,為源匯等地區(qū)提供建站服務(wù)。全流程按需網(wǎng)站制作，專業(yè)設(shè)計(jì)，全程項(xiàng)目跟蹤，創(chuàng)新互聯(lián)公司專業(yè)和態(tài)度為您提供的服務(wù)

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

kafka3.3.1新版本就會(huì)提示不兼容，需要改用新版本

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

創(chuàng)建多分區(qū) 副本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

可以進(jìn)入kafka的數(shù)據(jù)文件存儲(chǔ)目錄查看test和test1主題的消息日志文件：默認(rèn)log目錄 /tmp/kafka-logs

消息日志文件主要存放在分區(qū)文件夾里的以log結(jié)尾的日志文件里，如下是test-1主題對(duì)應(yīng)的分區(qū)0的消息日志：

leader節(jié)點(diǎn)負(fù)責(zé)給定partition的所有讀寫(xiě)請(qǐng)求。
replicas 表示某個(gè)partition在哪幾個(gè)broker上存在備份。不管這個(gè)幾點(diǎn)是不是”leader“，甚至這個(gè)節(jié)點(diǎn)掛了，也會(huì)列出。
isr 是replicas的一個(gè)子集，它只列出當(dāng)前還存活著的，并且已同步備份了該partition的節(jié)點(diǎn)。

刪除主題

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

更新主題

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

這個(gè)partition分區(qū)只能是增加分區(qū)，不可以降低，否則報(bào)錯(cuò)

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.

參數(shù)	描述
–bootstrap-server	連接的 Kafka Broker 主機(jī)名稱和端口號(hào)。
–topic	操作的 topic 名稱。
–create	創(chuàng)建主題。
–delete	刪除主題。
–alter	修改主題
–list	查看所有主題
–describe	查看主題詳細(xì)描述
–partitions	設(shè)置分區(qū)數(shù)
–replication-factor	設(shè)置分區(qū)副本
–config	更新系統(tǒng)默認(rèn)的配置。

producer發(fā)送消息

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

參數(shù)	描述
–bootstrap-server	連接的 Kafka Broker 主機(jī)名稱和端口號(hào)
–topic	操作的topic

consumer消費(fèi)消息

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

如果想要消費(fèi)之前的消息可以通過(guò)–from-beginning參數(shù)指定，如下命令：

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

消費(fèi)多主題

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

同一個(gè)消費(fèi)組下的消費(fèi)者，只能有一個(gè)消費(fèi)者消費(fèi)到topic下的消息

設(shè)置消費(fèi)組消息消息，單播消費(fèi)

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

多播消費(fèi)，如消費(fèi)訂閱，不同消費(fèi)組下的消費(fèi)者可以消費(fèi)同一topic下的同一條消息

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

列出消費(fèi)組列表

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

查看消費(fèi)組，消費(fèi)信息的消費(fèi)偏移信息

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

current-offset：當(dāng)前消費(fèi)組的已消費(fèi)偏移量
log-end-offset：主題對(duì)應(yīng)分區(qū)消息的結(jié)束偏移量(HW)
lag：當(dāng)前消費(fèi)組未消費(fèi)的消息數(shù)

參數(shù)	描述
–bootstrap-server	連接的 Kafka Broker 主機(jī)名稱和端口號(hào)
–topic	操作的 topic 名稱
–from-beginning	從頭開(kāi)始消費(fèi)
–group	指定消費(fèi)者組名稱。

你是否還在尋找穩(wěn)定的海外服務(wù)器提供商？創(chuàng)新互聯(lián)www.cdcxhl.cn海外機(jī)房具備T級(jí)流量清洗系統(tǒng)配攻擊溯源，準(zhǔn)確流量調(diào)度確保服務(wù)器高可用性，企業(yè)級(jí)服務(wù)器適合批量采購(gòu)，新人活動(dòng)首月15元起，快前往官網(wǎng)查看詳情吧

網(wǎng)頁(yè)名稱：Kafka-topicproducerconsumer常用命令-創(chuàng)新互聯(lián)
轉(zhuǎn)載來(lái)源：http://weahome.cn/article/dhedgo.html