快捷導航

kafka-console-consumer.sh使用2次grep管道無法提取消息的解決

更新時間：2023年03月07日 09:57:40 作者：Geoffrey Turing

這篇文章主要介紹了kafka-console-consumer.sh使用2次grep管道無法提取消息的解決方案，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教

問題

在使用kafka自帶腳本消費時，想用grep關(guān)鍵詞過濾出來想要的信息

./kafka-console-consumer.sh \
--bootstrap-server node1:9092,node2:9092,node3:9092 \
--topic test \
--from-beginning \
--consumer-property group.id=gaofeng_test\
 | grep  "b786aba6f3a6" | grep 388aabd

但是發(fā)現(xiàn)grep命令在這里有問題，無法取到想要消息

在這里插入圖片描述

已經(jīng)消費到latest了，都沒有提取到想要的msg

這里確認了producer端已經(jīng)發(fā)送成功了，回調(diào)函數(shù)里都沒有任何異常

懷疑

懷疑是否kafka存在丟數(shù)據(jù)的情況

排查問題后發(fā)現(xiàn)kafka集群一切正常，性能也很強勁

并且在存儲消息的日志文件中也已經(jīng)查詢到了我想要的消息了，為什么就是消費不出來呢

分析

將kafka的這個主題 --from-beginning 消費到文件topic.log中去

對topic.log這個靜態(tài)文件進行g(shù)rep “b786aba6f3a6” | grep 388aabd抽取，發(fā)現(xiàn)取到了自己想要的這條消息

kafka消費時結(jié)果是動態(tài)的，grep這里用法不對，只有對靜態(tài)的文件可以抽取到數(shù)據(jù)

--line-buffered
Force output to be line buffered. By default, output is line buffered when standard output is
a terminal and block buffered otherwise

上面的意思是

強制輸出結(jié)果使用行緩沖
默認情況下，如果標準輸入時終端，則使用line bufferred
否則，使用塊緩沖，（默認的大小為4096 bytes，因系統(tǒng)和配置而異）

所以，這也就解釋了為什么雙重grep過濾沒有內(nèi)容，因為沒有達到塊緩沖限制。

驗證

生產(chǎn)消息

a,b,c

在這里插入圖片描述

消費時兩次grep則無消息

在這里插入圖片描述

消費時一次grep則都出來

在這里插入圖片描述

結(jié)論

剛剛/kafka-console-consumer.sh第一次grep過濾出來的的消息小于4k了，所以傳給第二次grep是不可以直接輸出出來的，因為在緩沖區(qū)
如果第一次grep出來的數(shù)據(jù)大于4k，傳給第二個grep時則不會緩沖著，第二個grep收到就立刻傳入到terminal輸出了
反之第一次grep過濾出來的的消息大于還是小于4k，一旦后面沒有管道符號，直接輸出給terminal的話，則遵循line bufferred規(guī)則立馬輸出

解決

使用如下命令即可

./kafka-console-consumer.sh \
--bootstrap-server node1:9092,node2:9092,node3:9092 \
--topic test \
--from-beginning \
--consumer-property group.id=gaofeng_test\
 | grep --line-buffered "b786aba6f3a6" | grep 388aabd

【注意】

這里用了grep 的 --line-buffered參數(shù)

具體用法

查閱http://chabaoo.cn/article/277299.htm