Kafka常用命令之kafka-console-consumer.sh解讀

更新時(shí)間：2023年03月07日 10:08:33 作者：Ernest.Wu

這篇文章主要介紹了Kafka常用命令之kafka-console-consumer.sh解讀，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教

kafka-console-consumer.sh解讀

kafka-console-consumer.sh 腳本是一個(gè)簡(jiǎn)易的消費(fèi)者控制臺(tái)。

該 shell 腳本的功能通過(guò)調(diào)用 kafka.tools 包下的 ConsoleConsumer 類，并將提供的命令行參數(shù)全部傳給該類實(shí)現(xiàn)。

注意：Kafka 從 2.2 版本開(kāi)始將 kafka-topic.sh 腳本中的 −−zookeeper 參數(shù)標(biāo)注為 “過(guò)時(shí)”，推薦使用 −−bootstrap-server 參數(shù)。
若讀者依舊使用的是 2.1 及以下版本，請(qǐng)將下述的 --bootstrap-server 參數(shù)及其值手動(dòng)替換為 --zookeeper zk1:2181,zk2:2181,zk:2181。
一定要注意兩者參數(shù)值所指向的集群地址是不同的。

消息消費(fèi)

bin/kafka-console-consumer.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --topic topicName

表示從 latest 位移位置開(kāi)始消費(fèi)該主題的所有分區(qū)消息，即僅消費(fèi)正在寫(xiě)入的消息。

從開(kāi)始位置消費(fèi)

bin/kafka-console-consumer.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --from-beginning --topic topicName

表示從指定主題中有效的起始位移位置開(kāi)始消費(fèi)所有分區(qū)的消息。

顯示key消費(fèi)

bin/kafka-console-consumer.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --property print.key=true --topic topicName

消費(fèi)出的消息結(jié)果將打印出消息體的 key 和 value。

若還需要為你的消息添加其他屬性

請(qǐng)參考下述列表

參數(shù)	值類型	說(shuō)明	有效值
--topic	string	被消費(fèi)的topic
--whitelist	string	正則表達(dá)式，指定要包含以供使用的主題的白名單
--partition	integer	指定分區(qū) 除非指定’–offset’，否則從分區(qū)結(jié)束(latest)開(kāi)始消費(fèi)
--offset	string	執(zhí)行消費(fèi)的起始o(jì)ffset位置默認(rèn)值:latest	latest earliest <offset>
--consumer-property	string	將用戶定義的屬性以key=value的形式傳遞給使用者
--consumer.config	string	消費(fèi)者配置屬性文件請(qǐng)注意，[consumer-property]優(yōu)先于此配置
--formatter	string	用于格式化kafka消息以供顯示的類的名稱默認(rèn)值:kafka.tools.DefaultMessageFormatter	kafka.tools.DefaultMessageFormatter kafka.tools.LoggingMessageFormatter kafka.tools.NoOpMessageFormatter kafka.tools.ChecksumMessageFormatter
--property	string	初始化消息格式化程序的屬性	print.timestamp=true\|false print.key=true\|false print.value=true\|false key.separator=<key.separator> line.separator=<line.separator> key.deserializer=<key.deserializer> value.deserializer=<value.deserializer>
--from-beginning		從存在的最早消息開(kāi)始，而不是從最新消息開(kāi)始
--max-messages	integer	消費(fèi)的最大數(shù)據(jù)量，若不指定，則持續(xù)消費(fèi)下去
--timeout-ms	integer	在指定時(shí)間間隔內(nèi)沒(méi)有消息可用時(shí)退出
--skip-message-on-error		如果處理消息時(shí)出錯(cuò)，請(qǐng)?zhí)^(guò)它而不是暫停
--bootstrap-server	string	必需(除非使用舊版本的消費(fèi)者)，要連接的服務(wù)器
--key-deserializer	string
--value-deserializer	string
--enable-systest-events		除記錄消費(fèi)的消息外，還記錄消費(fèi)者的生命周期 (用于系統(tǒng)測(cè)試)
--isolation-level	string	設(shè)置為read_committed以過(guò)濾掉未提交的事務(wù)性消息設(shè)置為read_uncommitted以讀取所有消息默認(rèn)值:read_uncommitted
--group	string	指定消費(fèi)者所屬組的ID
--blacklist	string	要從消費(fèi)中排除的主題黑名單
--csv-reporter-enabled		如果設(shè)置，將啟用csv metrics報(bào)告器
--delete-consumer-offsets		如果指定，則啟動(dòng)時(shí)刪除zookeeper中的消費(fèi)者信息
--metrics-dir	string	輸出csv度量值需與[csv-reporter-enable]配合使用
--zookeeper	string	必需(僅當(dāng)使用舊的使用者時(shí))連接zookeeper的字符串。可以給出多個(gè)URL以允許故障轉(zhuǎn)移