日韩成人免费在线_国产成人一二_精品国产免费人成电影在线观..._日本一区二区三区久久久久久久久不

當(dāng)前位置:首頁 > 科技  > 軟件

Spring實(shí)現(xiàn)Kafka重試Topic,真的太香了

來源: 責(zé)編: 時(shí)間:2024-01-26 09:02:59 226觀看
導(dǎo)讀概述Kafka的強(qiáng)大功能之一是每個(gè)分區(qū)都有一個(gè)Consumer的偏移值。該偏移值是消費(fèi)者將讀取的下一條消息的值。可以自動或手動增加該值。如果我們由于錯(cuò)誤而無法處理消息并想重試,我們可以選擇手動管理,并在成功的情況下增

概述

Kafka的強(qiáng)大功能之一是每個(gè)分區(qū)都有一個(gè)Consumer的偏移值。該偏移值是消費(fèi)者將讀取的下一條消息的值。可以自動或手動增加該值。如果我們由于錯(cuò)誤而無法處理消息并想重試,我們可以選擇手動管理,并在成功的情況下增加偏移量。但是,這會暫時(shí)阻止隊(duì)列消息的處理。我們可以選擇異步方法。2wI28資訊網(wǎng)——每日最新資訊28at.com

為什么我們需要它?

如果發(fā)生錯(cuò)誤,而不是停止隊(duì)列消息的處理;我們可以將錯(cuò)誤消息轉(zhuǎn)移到不同的主題并再次處理。2wI28資訊網(wǎng)——每日最新資訊28at.com

如果在處理 Kafka 消息時(shí)出現(xiàn)錯(cuò)誤,可以使用 RetryableTopic 注解以一定的時(shí)間間隔和一定的次數(shù)再次處理消息。如果完成嘗試次數(shù)后錯(cuò)誤仍然存在,則消息將發(fā)送到 DLT 隊(duì)列。2wI28資訊網(wǎng)——每日最新資訊28at.com

如何使用?

我們首先回顧一下RetryableTopic注解可以取的一些值,以便您可以做出最適合您的設(shè)置:2wI28資訊網(wǎng)——每日最新資訊28at.com

attempts:嘗試處理消息的次數(shù)。它的默認(rèn)值為 3。如果完成所有嘗試后仍然收到錯(cuò)誤,則消息將發(fā)送到 DLT 隊(duì)列。2wI28資訊網(wǎng)——每日最新資訊28at.com

backoff:用于確定處理消息的時(shí)間間隔。從 Backoff 類獲取一個(gè)值。您可以在下面找到退避的詳細(xì)示例。2wI28資訊網(wǎng)——每日最新資訊28at.com

排除/排除名稱:允許您排除指定的異常類。當(dāng)您添加到列表中的任何錯(cuò)誤被拋出時(shí),重試機(jī)制將不會被激活。2wI28資訊網(wǎng)——每日最新資訊28at.com

include / includeNames:僅當(dāng)拋出指定的異常時(shí)才會激活重試機(jī)制。2wI28資訊網(wǎng)——每日最新資訊28at.com

kafkaTemplate:雖然您可以給出現(xiàn)有 kafkaTemplate bean 的名稱,但您也可以為特定于重試的 Kafka 模板定義不同的 bean。2wI28資訊網(wǎng)——每日最新資訊28at.com

autoCreateTopics:決定是否自動創(chuàng)建Retry和DLT主題。2wI28資訊網(wǎng)——每日最新資訊28at.com

retryTopicSuffix / dltTopicSuffix:用于確定要添加到自動創(chuàng)建的主題末尾的后綴。2wI28資訊網(wǎng)——每日最新資訊28at.com

dltStrategy:如果不需要DLT,可以定義為NO_DLT。2wI28資訊網(wǎng)——每日最新資訊28at.com

SameIntervalTopicReuseStrategy/fixedDelayTopicStrategy(3.0.4之前):用于確定要?jiǎng)?chuàng)建的重試主題策略。創(chuàng)建 (SINGLE_TOPIC) 或盡可能多的嘗試值 (MULTIPLE_TOPICS) 重試主題。2wI28資訊網(wǎng)——每日最新資訊28at.com

Backoff的示例:

  • 具有固定的增量值
Backoff(delay = 600000 ) // 每 10 分鐘
  • 具有指數(shù)價(jià)值
Backoff(delay = 60000 , multiplier = 2 ) // 1、2、4、8... 分鐘后重復(fù)。
  • 用占位符定義值
Backoff(delayExpression = "${delay}", multiplierExpression = "${multiplier}")

@RetryableTopic 示例:

@RetryableTopic(     backoff = @Backoff(delay = 300000),     attempts = 12,     sameIntervalTopicReuseStrategy =          SameIntervalTopicReuseStrategy.SINGLE_TOPIC,     kafkaTemplate = "kafkaRetryableTopicTemplate",     exclude = { SerializationException.class,                  DeserializationException.class,                  NullPointerException.class                } ) @KafkaListener(topics = "my-topic") public void processMessage(RetryableDto retryableDto) {     log.info("Retrying process RetryableDto : {}", retryableDto);     // process message }

在上面的例子中,消息將每5分鐘重新處理一次,總共12次,即1小時(shí)。如果任何嘗試均順利完成,則試用將終止。2wI28資訊網(wǎng)——每日最新資訊28at.com

由于定義了 SINGLE_TOPIC,因此將創(chuàng)建單個(gè)主題以進(jìn)行重試。如果沒有進(jìn)行此定義,則會創(chuàng)建 12 個(gè)重試主題。2wI28資訊網(wǎng)——每日最新資訊28at.com

如果拋出了排除中定義的任何錯(cuò)誤,則不會執(zhí)行重做。2wI28資訊網(wǎng)——每日最新資訊28at.com

如果需要,您可以編寫自己的 RetryableException 并在包含中定義此值,以便僅在引發(fā)此錯(cuò)誤時(shí)才重試。2wI28資訊網(wǎng)——每日最新資訊28at.com

DLT隊(duì)列處理

如果完成了定義的嘗試次數(shù)并且繼續(xù)收到錯(cuò)誤,則消息將發(fā)送到 DLT 隊(duì)列。如果要處理這些消息,可以使用DltHandler注解。2wI28資訊網(wǎng)——每日最新資訊28at.com

用法示例:2wI28資訊網(wǎng)——每日最新資訊28at.com

@DltHandler  public  void  handleDltMessage (RetryableDto retryableDto) {      log.error("DLT處理程序消息:{}", retryableDto); }

注意事項(xiàng)

雖然使用 RetryableTopic 的異步處理優(yōu)勢為我們帶來了性能提升,但這種使用也有一些缺點(diǎn)。2wI28資訊網(wǎng)——每日最新資訊28at.com

使用RetryableTopic可能會破壞消息的處理順序。2wI28資訊網(wǎng)——每日最新資訊28at.com

讓我們用一個(gè)例子來解釋這種情況:當(dāng)主主題在時(shí)間 t 處理時(shí),一條消息出錯(cuò)并被發(fā)送到重試主題。在時(shí)間 t + 1 時(shí),另一條消息來到主主題并成功處理。讓我們在重試主題中的消息在時(shí)間 t + 2 時(shí)被成功處理。在這種情況下,第一條傳入消息將在第二條消息之后處理。如果訂購對您很重要,我建議您在消息處理過程中進(jìn)行必要的檢查。2wI28資訊網(wǎng)——每日最新資訊28at.com

另一個(gè)缺點(diǎn)是消息雙重處理的風(fēng)險(xiǎn)。您可以通過考慮這種可能性來進(jìn)行改進(jìn)。2wI28資訊網(wǎng)——每日最新資訊28at.com

本文鏈接:http://www.www897cc.com/showinfo-26-68327-0.htmlSpring實(shí)現(xiàn)Kafka重試Topic,真的太香了

聲明:本網(wǎng)頁內(nèi)容旨在傳播知識,若有侵權(quán)等問題請及時(shí)與本網(wǎng)聯(lián)系,我們將在第一時(shí)間刪除處理。郵件:2376512515@qq.com

上一篇: 如何使用Python、Apache Kafka和云平臺構(gòu)建健壯的實(shí)時(shí)數(shù)據(jù)管道

下一篇: 性能篇:解密Stream,提升集合遍歷效率的秘訣!

標(biāo)簽:
  • 熱門焦點(diǎn)
Top 主站蜘蛛池模板: 永和县| 从江县| 腾冲县| 阿瓦提县| 漾濞| 上饶市| 江油市| 西峡县| 新建县| 鞍山市| 乌兰浩特市| 松江区| 宣城市| 青冈县| 新化县| 库伦旗| 万源市| 天镇县| 陈巴尔虎旗| 清苑县| 仁化县| 友谊县| 博罗县| 镇远县| 遵义市| 富顺县| 海原县| 阿克苏市| 东城区| 静乐县| 沭阳县| 济阳县| 宜都市| 微山县| 临颍县| 丰镇市| 远安县| 墨玉县| 常山县| 绿春县| 永和县|