快捷導(dǎo)航

kafka監(jiān)聽問題的解決和剖析

更新時(shí)間：2020年12月03日 11:31:42 作者：知曉汝名,嚇吾一跳！

這篇文章主要給大家介紹了關(guān)于kafka監(jiān)聽問題的解決和剖析的相關(guān)資料，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

問題如下：
一. 解決問題一（kafka監(jiān)聽不到數(shù)據(jù)）
二. 解決問題二（kafka為什么會(huì)有重復(fù)數(shù)據(jù)發(fā)送）
三. 解決問題三（kafka數(shù)據(jù)重復(fù)如何解決）
四. 解決問題四（為什么kafka會(huì)出現(xiàn)倆個(gè)消費(fèi)端都可以消費(fèi)問題）
五. 粘一下我的監(jiān)聽配置文件
總結(jié)

問題如下：

kafka為什么監(jiān)聽不到數(shù)據(jù)
kafka為什么會(huì)有重復(fù)數(shù)據(jù)發(fā)送
kafka數(shù)據(jù)重復(fù)如何解決
為什么kafka會(huì)出現(xiàn)倆個(gè)消費(fèi)端都可以消費(fèi)問題
kafka監(jiān)聽配置文件

一. 解決問題一（kafka監(jiān)聽不到數(shù)據(jù)）

　　首先kafka監(jiān)聽不得到數(shù)據(jù),檢查如下

檢查配置文件是否正確（可能會(huì)出現(xiàn)改了監(jiān)聽地址,監(jiān)聽Topic,監(jiān)聽的地址的數(shù)量問題）
檢查接收數(shù)據(jù)的正確性（比如原生的代碼,可能是用byte序列化接收的數(shù)據(jù),而你接收使用String。也是配置文件序列化問題,還有與發(fā)送者商量問題）
檢查kafka版本問題（一般的版本其實(shí)是沒什么問題的,只有個(gè)別版本會(huì)出現(xiàn)監(jiān)聽不到問題）
沒有加
@Component 犯了最不應(yīng)該出差錯(cuò)的問題

　　如果出現(xiàn)監(jiān)聽不到數(shù)據(jù)的問題,那么就試試更改方法一二,如果不可以在去試試方法三,之前出現(xiàn)這個(gè)問題也是查過一般查到都會(huì)說 “低版本的服務(wù)器接收不到高版本的生產(chǎn)者發(fā)送的消息”,但是凈由測試使用用1.0.5RELEASE 和 2.6.3反復(fù)測試,并沒有任何的問題。

如果按照版本一致,那么根本就不現(xiàn)實(shí),因?yàn)榭赡懿煌捻?xiàng)目,springboot版本不一致的話,可能有的springboot版本低,那么你還得要求自己維護(hù)項(xiàng)目版本升級(jí)?如果出現(xiàn)第四種情況就無話可說了。

二. 解決問題二（kafka為什么會(huì)有重復(fù)數(shù)據(jù)發(fā)送）

　　重復(fù)數(shù)據(jù)的發(fā)送問題如下

可能在發(fā)送者的那里的事務(wù)問題。mysql存儲(chǔ)事務(wù)發(fā)生異常導(dǎo)致回滾操作,但是kafka消息卻是已經(jīng)發(fā)送到了服務(wù)器中。此事肯定會(huì)出現(xiàn)重復(fù)問題
生產(chǎn)者設(shè)置時(shí)間問題,生產(chǎn)發(fā)送設(shè)置的時(shí)間內(nèi),消息沒完成發(fā)送,生產(chǎn)者以為消費(fèi)者掛掉,便重新發(fā)送一個(gè),導(dǎo)致重復(fù)
offset問題,當(dāng)項(xiàng)目重啟,offset走到某一個(gè)位置已扔到kafka服務(wù)器中,但是項(xiàng)目被重啟.那么offset會(huì)是在原本重啟的那一個(gè)點(diǎn)的地方再次發(fā)送一次,這是kafka設(shè)計(jì)的問題,防止出現(xiàn)丟失數(shù)據(jù)問題

三. 解決問題三（kafka數(shù)據(jù)重復(fù)如何解決）

　　目前我是使用的Redis進(jìn)行的排重法,用的是Redis中的set,保證里面不存在重復(fù),保證Redis里面不會(huì)存入太多的臟數(shù)據(jù)。并定期清理

　　粘貼一下我的排重（Redis排重法）

//kafka prefix
  String cache = "kafka_cache";
  //kafka suffix
  Calendar c = Calendar.getInstance();
  SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  //0點(diǎn),目前是為了設(shè)置為這一天的固定時(shí)間。這個(gè)完全可以去寫個(gè)工具類自己弄,為了看的更清楚,麻煩了一點(diǎn)的寫入
  SimpleDateFormat sdf2 = new SimpleDateFormat("yyyy-MM-dd 00:00:00");
  String gtimeStart = sdf2.format(c.getTime());
  long time = sdf.parse(gtimeStart).getTime();


  //此位置為了設(shè)置是否是新的一天，新的一天需要設(shè)置定時(shí)時(shí)間,保證redis中不會(huì)存儲(chǔ)太多無用數(shù)據(jù)
  Boolean flag = false;
  //數(shù)據(jù)接收
  Set<String> range = new HashSet<>();
  //判斷是否存在
  if (redisTemplate.hasKey(cache + time)) {
  //存在則取出這個(gè)set
  range = redisTemplate.opsForSet().members(cache + time);
  }else {
  //不存在,則為下面過期時(shí)間的設(shè)置鋪墊
  flag = true;
  }
  //判斷監(jiān)聽到的數(shù)據(jù)是否是重復(fù)
  if (range.contains("測試需要")) {
  //重復(fù)則排出,根據(jù)邏輯自己修改
  continue;
  } else {
  //添加進(jìn)去
  redisTemplate.opsForSet().add(cache + time, i+"");
  if (flag){
   //設(shè)置為24小時(shí),保證新一天使用,之前使用的存儲(chǔ)會(huì)消失掉
   redisTemplate.expire(cache + time,24,TimeUnit.HOURS);
   //不會(huì)在進(jìn)入這個(gè)里面,如果多次的存入過期時(shí)間,那么這個(gè)key的過期時(shí)間就永遠(yuǎn)是24小時(shí),一直就不會(huì)過期
   flag = false;
  }
  }

四. 解決問題四（為什么kafka會(huì)出現(xiàn)倆個(gè)消費(fèi)端都可以消費(fèi)問題）

　　原因是因?yàn)樵诓煌琯roupId之下,kafka接收到以后,會(huì)給監(jiān)聽他的每一個(gè)組發(fā)送一個(gè)他所收到的消息,但是兩個(gè)消費(fèi)端監(jiān)聽同一個(gè)租,那么就只有一個(gè)消費(fèi)端可以消費(fèi)到。

五. 粘一下我的監(jiān)聽配置文件

# 指定kafka 代理地址，可以多個(gè),用逗號(hào)間隔
spring.kafka.bootstrap-servers= localhost:9092
# 指定默認(rèn)消費(fèi)者group id
spring.kafka.consumer.group-id= test
# 是否自動(dòng)提交
spring.kafka.consumer.enable-auto-commit= true
# 提交間隔的毫秒
spring.kafka.consumer.auto-commit-interval.ms=60000
# 最大輪詢的次數(shù)
spring.kafka.consumer.max-poll-records=1
# 將偏移量重置為最新偏移量
spring.kafka.consumer.auto-offset-reset=earliest
# 指定消息key和消息體的編解碼方式
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer