本文介紹了如何避免使用Kafka流丟失消息的處理方法，對大家解決問題具有一定的參考價值，需要的朋友們下面隨著小編來一起學習吧！
問題描述
我們有一個Streams應用程序，它使用源主題中的消息，執行一些處理并將結果轉發到目標主題。
消息的結構由某些Avro架構控制。
當開始使用消息時，如果架構尚未緩存，應用程序將嘗試從架構注冊表中檢索它。如果由于任何原因架構注冊表不可用(例如網絡故障)，則當前正在處理的消息將丟失，因為默認處理程序是名為LogAndContinueExceptionHandler的處理程序。
o.a.k.s.e.LogAndContinueExceptionHandler : Exception caught during Deserialization, taskId: 1_5, topic: my.topic.v1, partition: 5, offset: 142768
org.apache.kafka.common.errors.SerializationException: Error retrieving Avro schema for id 62
Caused by: java.net.SocketTimeoutException: connect timed out
at java.base/java.net.PlainSocketImpl.socketConnect(Native Method) ~[na:na]
...
o.a.k.s.p.internals.RecordDeserializer : stream-thread [my-app-StreamThread-3] task [1_5] Skipping record due to deserialization error. topic=[my.topic.v1] partition=[5] offset=[142768]
...
因此，我的問題是，處理上述情況的正確方法是什么，并確保無論如何都不會丟失消息。是否有現成的LogAndRollbackExceptionHandler錯誤處理程序或實現您自己的方法？
提前感謝您的投入。
推薦答案
我在Kafka上的工作不是很多，但當我工作時，我記得遇到了您在我們系統中描述的問題。
讓我告訴你我們是如何處理我們的場景的，也許這也會對你有所幫助：
場景1：如果您的消息在發布方(Publisher–&>Kafka)丟失，您可以根據需要配置Kafka確認設置，如果您使用的是SpringCloud Stream和Kafka，則該屬性為spring.cloud.stream.kafka.binder.required-acks
可能的值：
最多一次(Ack=0)
-
出版商不在乎卡夫卡是否承認。
發送并忘記
可能會丟失數據
至少一次(Ack=1)
如果Kafka不確認，出版商將重新發送消息。
可能存在重復。
在將郵件復制到副本之前發送確認。
正好一次(Ack=All)
如果Kafka不確認，出版商將重新發送消息。
但是，如果一條消息多次發送到Kafka，則不會有重復。
內部序列號，用于判斷消息是否已經寫入主題。
需要設置Min.insync.Replicas屬性，以確保在Kafka向生產者確認之前，需要同步的最小復制數是多少。
場景2：如果您的數據在消費者端丟失(Kafka–&>Consumer)，您可以根據您的使用情況更改Kafka的自動提交功能。如果您使用的是Spring Cloud Streamspring.cloud.stream.kafka.bindings.input.consumer.AutoCommitOffset.
，則為該屬性
默認情況下，在Kafka中AutoCommittee Offset為True，并且發送給消費者的每一條消息在Kafka的末尾都是&Quot；Submitted&Quot；，這意味著它不會被再次發送。但是，如果您將AutoCommittee Offset更改為False，您將有權在代碼中輪詢來自Kafka的消息，并且一旦您完成工作，將Commit顯式設置為True，以讓Kafka知道您現在已經完成了消息。
如果消息未提交，Kafka將繼續重新發送，直到消息提交為止。
希望這對您有所幫助，或至少為您指明正確的方向。
這篇關于如何避免使用Kafka流丟失消息的文章就介紹到這了，希望我們推薦的答案對大家有所幫助，
