如何在項目中使用spring-kafka操作消費kafka？

如何在項目中使用spring-kafka操作消費kafka？針對這個問題，今天小編總結這篇有關kafka的文章，希望能幫助更多想解決這個問題的朋友找到更加簡單易行的辦法。

日志分析

以下為日志記錄，consumer會時長從消費組中掉出再重新加入

o.a.k.c.c.i.AbstractCoordinator:542 - Marking the coordinator 127.0.0.1:9092 (id: 2147483643 rack: null) dead for group group.id 
o.a.k.c.c.i.AbstractCoordinator:505 - Discovered coordinator 127.0.0.1:9092 (id: 2147483643 rack: null) for group group.id. 
o.a.k.c.c.i.ConsumerCoordinator:292 - Revoking previously assigned partitions [topic-12] for group group.id 
o.a.k.c.c.i.AbstractCoordinator:326 - (Re-)joining group group.id 
o.a.k.c.c.i.AbstractCoordinator$SyncGroupResponseHandler:434 - Successfully joined group group.id with generation 12182 
o.a.k.c.c.i.ConsumerCoordinator:231 - Setting newly assigned partitions [topic-12] for group group.id 

解決方案

kafka服務配置文件server.properties提高系統(tǒng)配置參數(shù)

#Broker處理消息的最大線程數(shù)
num.network.threads=cpu核數(shù)加1
#Broker處理磁盤IO的線程數(shù)
num.io.threads=cpu核數(shù)2倍，最大不超過3倍

備份并重啟kafka服務

以上就是在項目中使用spring-kafka操作消費kafka的方法介紹，詳細使用情況還得要大家自己使用過才能知道具體要領。如果想閱讀更多相關內容的文章，歡迎關注億速云行業(yè)資訊頻道！