kafka消息丟失怎么辦

本篇內容主要講解“kafka消息丟失怎么辦”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學習“kafka消息丟失怎么辦”吧!

在處理生產環境問題的過程中發現12號那天某kafka集群有少量的數據丟失，概率大致在千萬分之三。 數據寫入kafka之后，就完全消失了，消費者完全沒有消費到這個數據。 通過找到那天的數據，查看有問題的數據在寫入kafka的時候上下文應用日志發現有少量以下報錯:

[2019-10-12 11:03:43,xxx] This is not the correct coordinator.

理論上正常情況下kafka是不太可能丟數據的，如果出現這種情況，必然是開發人員或者硬件引發了什么問題，因為寫入日志是有的，看了下應用配置

acks=1

馬上意識到，問題突破口應該在這里。

acks=0 生產者能夠通過網絡吧消息發送出去，那么就認為消息已成功寫入Kafka,一定會丟失一些數據
acks=1 master在疏導消息并把它寫到分區數據問津是會返回確認或者錯誤響應，還是可能會丟數據
acks=all master在返回確認或錯誤響應之前，會等待所有同步副本都收到消息

可能以前是為了保證性能夠快，選擇了折中的應用配置 acks=1 。

馬上想到去看下kafka的日志，猜測這個時間段必然出現出現了 master 不可用的情況才會導致數據丟失。

在kafka集群的57號節點的機器上看到這樣一段日志：

[2019-10-12 11:03:39,427] WARN Client session timed out, have not heard from server in 4034ms for sessionid 0x396aaaadbbxx00 (org.apache.zookeeper.ClientCnxn)
[2019-10-12 11:03:40,908] INFO Client session timed out, have not heard from server in 4034ms for sessionid 0x396aaaabbxx00, closing socket connection and attempting reconnect (org.apache.zookeeper.ClientCnxn)
[2019-10-12 11:03:41,253] INFO zookeeper state changed (Disconnected) (org.I0Itec.zkclient.ZkClient)
[2019-10-12 11:03:41,962] INFO Opening socket connection to server xx.xx.xx.59/xx.xx.xx.59:2181. Will not attempt to authenticate using SASL (unknown error) (org.apache.zookeeper.ClientCnxn)
[2019-10-12 11:03:41,962] INFO Socket connection established to xx.xx.xx.59/10.xx.xx.xx:2181, initiating session (org.apache.zookeeper.ClientCnxn)
[2019-10-12 11:03:42,293] WARN Unable to reconnect to ZooKeeper service, session 0x396cf664cdbb0000 has expired (org.apache.zookeeper.ClientCnxn)
[2019-10-12 11:03:42,293] INFO zookeeper state changed (Expired) (org.I0Itec.zkclient.ZkClient)
[2019-10-12 11:03:42,294] INFO Initiating client connection, connectString=xx.xx.xx.55:2181,xx.xx.xx.56:2181,xx.xx.xx.57:2181,xx.xx.xx.58:2181,xx.xx.xx.59:2181 sessionTimeout=6000 watcher=org.I0Itec.zkclient.ZkClient@342xxx2d (org.apache.zookeeper.ZooKeeper)
[2019-10-12 11:03:42,313] INFO Unable to reconnect to ZooKeeper service, session 0x396cxxxxxb0000 has expired, closing socket connection (org.apache.zookeeper.ClientCnxn)
[2019-10-12 11:03:42,323] INFO EventThread shut down for session: 0x396cxxxx000 (org.apache.zookeeper.ClientCnxn)
[2019-10-12 11:03:42,342] INFO Opening socket connection to server xx.xx.xx.58/xx.xx.xx.58:2181. Will not attempt to authenticate using SASL (unknown error) (org.apache.zookeeper.ClientCnxn)
[2019-10-12 11:03:42,343] INFO Socket connection established to xx.xx.xx.58/xx.xx.xx.58:2181, initiating session (org.apache.zookeeper.ClientCnxn)
[2019-10-12 11:03:43,516] INFO Session establishment complete on server xx.xx.xx.58/xx.xx.xx.58:2181, sessionid = 0x3ax4xxxxx01, negotiated timeout = 6000 (org.apache.zookeeper.ClientCnxn)
[2019-10-12 11:03:43,517] INFO zookeeper state changed (SyncConnected) (org.I0Itec.zkclient.ZkClient)

看來和猜測的情況是一樣的，這個kafka的節點機器出現了連接不到zk，所以這臺機器處于‘丟失’狀態。又去看了下kafka的controller日志：

[2019-10-12 11:03:42,671] INFO [Controller id=56] Newly added brokers: , deleted brokers: 57, all live brokers: ...,55,56,58,59,xx.... (kafka.controller.KafkaController)
[2019-10-12 11:03:42,678] INFO [Controller-56-to-broker-57-send-thread]: Shutting down (kafka.controller.RequestSendThread)
[2019-10-12 11:03:42,749] INFO [Controller-56-to-broker-57-send-thread]: Stopped (kafka.controller.RequestSendThread)
[2019-10-12 11:03:42,749] INFO [Controller-56-to-broker-57-send-thread]: Shutdown completed (kafka.controller.RequestSendThread)
[2019-10-12 11:03:42,828] INFO [Controller id=56] Broker failure callback for 57 (kafka.controller.KafkaController)
[2019-10-12 11:03:42,836] INFO [Controller id=56] Removed ArrayBuffer() from list of shutting down brokers. (kafka.controller.KafkaController)

再次看了下這個節點系統的網絡層面監控：

真相浮出水面，和猜測吻合，這個問題和之前遇到的一個問題有點類似，不一樣的情況是這次只丟失了一個節點，上次的問題是所有的網絡節點都短暫的丟失，詳情可以看我之前寫的博客： https://my.oschina.net/110NotFound/blog/3105190

這次的原因分析：

57號節點丟失之后，立馬進行了選舉，這個時候數據在生產的時候到達了master，恰好選舉的時候變更了master，而 acks=1 ，數據已經進入master，但是還沒有來得及同步到slave，這樣就導致了數據的丟失。

總結：

如果有對數據有強一致性的要求，一定要選擇 acks=all 否則指不定哪天硬件的輕微系統抖動就會導致 kafka 集群重新選舉，丟失數據。

到此，相信大家對“kafka消息丟失怎么辦”有了更深的了解，不妨來實際操作一番吧！這里是億速云網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！