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

這篇文章主要講解了“kafka-console-consumer.sh使用2次grep管道無法提取消息如何解決”，文中的講解內容簡單清晰，易于學習與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學習“kafka-console-consumer.sh使用2次grep管道無法提取消息如何解決”吧！

問題

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

./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

但是發現grep命令在這里有問題，無法取到想要消息

已經消費到latest了，都沒有提取到想要的msg

這里確認了producer端已經發送成功了，回調函數里都沒有任何異常

懷疑

懷疑是否kafka存在丟數據的情況

排查問題后發現kafka集群一切正常，性能也很強勁

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

分析

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

對topic.log這個靜態文件進行grep “b786aba6f3a6” | grep 388aabd抽取，發現取到了自己想要的這條消息

kafka消費時結果是動態的，grep這里用法不對，只有對靜態的文件可以抽取到數據

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

上面的意思是

• 強制輸出結果使用行緩沖

• 默認情況下，如果標準輸入時終端，則使用line bufferred

• 否則，使用塊緩沖，（默認的大小為4096 bytes，因系統和配置而異）

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

驗證

生產消息

a,b,c

消費時兩次grep則無消息

消費時一次grep則都出來

結論

• 剛剛/kafka-console-consumer.sh第一次grep過濾出來的的消息小于4k了，所以傳給第二次grep是不可以直接輸出出來的，因為在緩沖區

• 如果第一次grep出來的數據大于4k，傳給第二個grep時則不會緩沖著，第二個grep收到就立刻傳入到terminal輸出了

• 反之第一次grep過濾出來的的消息大于還是小于4k，一旦后面沒有管道符號，直接輸出給terminal的話，則遵循line bufferred規則立馬輸出

解決

使用如下命令即可

./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參數

感謝各位的閱讀，以上就是“kafka-console-consumer.sh使用2次grep管道無法提取消息如何解決”的內容了，經過本文的學習后，相信大家對kafka-console-consumer.sh使用2次grep管道無法提取消息如何解決這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是億速云，小編將為大家推送更多相關知識點的文章，歡迎關注！