問題

在使用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

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

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

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

懷疑

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

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

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

分析

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

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

kafka消費時結果是動態(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

上面的意思是

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

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

驗證

生產消息

a,b,c

消費時兩次grep則無消息

消費時一次grep則都出來

結論

• 剛剛/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://www.dbjr.com.cn/article/277299.htm

最后

以上為個人經驗，希望能給大家一個參考，也希望大家多多支持腳本之家。

最新評論