Rabbit消息重試機(jī)制問題記錄
更新時(shí)間：2024年08月09日 10:21:25 作者：陳亦康
消息重試機(jī)制就是在消息處理失敗之后重新發(fā)送,主要時(shí)為了解決消息發(fā)送過程可能會(huì)出現(xiàn)的問題,例如 網(wǎng)絡(luò)故障、服務(wù)臨時(shí)不可用 等,這篇文章主要介紹了Rabbit消息重試機(jī)制問題記錄,需要的朋友可以參考下
消息重試機(jī)制
概述
消息重試機(jī)制就是在消息處理失敗之后重新發(fā)送，主要時(shí)為了解決消息發(fā)送過程可能會(huì)出現(xiàn)的問題，例如 網(wǎng)絡(luò)故障、服務(wù)臨時(shí)不可用 等.
Ps：如果時(shí)程序邏輯引起的錯(cuò)誤，那么即使重試多少次都是沒有用的，但是可以通過配置重試次數(shù)來解決.
實(shí)現(xiàn)方式一：基于消息手動(dòng)確認(rèn)機(jī)制，返回 nack 實(shí)現(xiàn)
配置文件
spring:
application:
name: rabbitmq
rabbitmq:
host: env-base
port: 5672
username: root
password: 1111
listener:
simple:
acknowledge-mode: manual # 手動(dòng)確認(rèn)交換機(jī)、隊(duì)列、綁定
@Bean("ackExchange")
fun ackExchange() = DirectExchange(MQConst.ACK_EXCHANGE)
@Bean("ackQueue")
fun ackQueue() = Queue(MQConst.ACK_QUEUE)
@Bean
fun ackBinding(
@Qualifier("ackExchange") exchange: DirectExchange,
@Qualifier("ackQueue") queue: Queue,
): Binding {
return BindingBuilder
.bind(queue)
.to(exchange)
.with(MQConst.ACK_BINDING)
}生產(chǎn)者接口
@RestController
@RequestMapping("/mq3")
class MQ3Api(
val rabbitTemplate: RabbitTemplate
) {
@RequestMapping("/ack")
fun ack(): String {
rabbitTemplate.convertAndSend(MQConst.ACK_EXCHANGE, MQConst.ACK_BINDING, "ack msg 1")
return "ok"
}
}消費(fèi)者
import com.cyk.rabbitmq.constants.MQConst
import com.rabbitmq.client.Channel //注意這里的依賴
import org.springframework.amqp.core.Message //注意這里的依賴
import org.springframework.amqp.rabbit.annotation.RabbitListener
import org.springframework.stereotype.Component
import java.nio.charset.Charset
@Component
class AckListener {
@RabbitListener(queues = [MQConst.ACK_QUEUE])
fun handMessage(
message: Message,
channel: Channel,
) {
val deliveryTag = message.messageProperties.deliveryTag
try {
println("接收到消息: ${String(message.body, Charset.forName("UTF-8"))}, $deliveryTag")
val a = 1 / 0
channel.basicAck(deliveryTag, false)
} catch (e: Exception) {
//通過返回 nack，并設(shè)置 requeue 為 ture 實(shí)現(xiàn)消息重新入隊(duì)，并進(jìn)行重試
channel.basicNack(deliveryTag, false, true)
}
}
}演示和結(jié)論
deliverTag 自增的原因： 引發(fā)異常后，會(huì)返回 nack，并且參數(shù) requeue = true，表示重新入隊(duì)，然后進(jìn)行重試，將隊(duì)列中的消息再次發(fā)送給生產(chǎn)者，因此 deliverTag 會(huì)自增.
缺點(diǎn)： 如果是由于程序邏輯異常引起的重試，那么無論重試多少次都沒用，并且不斷重試會(huì)導(dǎo)致負(fù)載飆升，性能下降.
實(shí)現(xiàn)方式二：基于重試配置實(shí)現(xiàn) 配置文件
spring:
application:
name: rabbitmq
rabbitmq:
host: env-base
port: 5672
username: root
password: 1111
listener:
simple:
acknowledge-mode: auto # 開啟重試機(jī)制，這里必須是 auto，否則不生效！
retry:
enabled: true # 開啟消費(fèi)者失敗重試
initial-interval: 5000ms # 失敗等待時(shí)常
max-attempts: 5 # 最大重試次數(shù)(包括第一次消費(fèi))Ps：開啟重試機(jī)制，acknowledge-mode 必須指定為 auto，否則不生效！
交換機(jī)、隊(duì)列、綁定
@Bean("ackExchange")
fun ackExchange() = DirectExchange(MQConst.ACK_EXCHANGE)
@Bean("ackQueue")
fun ackQueue() = Queue(MQConst.ACK_QUEUE)
@Bean
fun ackBinding(
@Qualifier("ackExchange") exchange: DirectExchange,
@Qualifier("ackQueue") queue: Queue,
): Binding {
return BindingBuilder
.bind(queue)
.to(exchange)
.with(MQConst.ACK_BINDING)
}生產(chǎn)者接口
@RequestMapping("/ack")
fun ack(): String {
rabbitTemplate.convertAndSend(MQConst.ACK_EXCHANGE, MQConst.ACK_BINDING, "ack msg 1")
return "ok"
}消費(fèi)者
import com.cyk.rabbitmq.constants.MQConst
import com.rabbitmq.client.Channel //注意這里的依賴
import org.springframework.amqp.core.Message //注意這里的依賴
import org.springframework.amqp.rabbit.annotation.RabbitListener
import org.springframework.stereotype.Component
import java.nio.charset.Charset
@Component
class AckListener {
@RabbitListener(queues = [MQConst.ACK_QUEUE])
fun handMessage(
message: Message,
channel: Channel,
) {
println("接收到消息: ${String(message.body, Charset.forName("UTF-8"))}, ${message.messageProperties.deliveryTag}")
val a = 1 / 0
}
}演示和結(jié)論
deliverTag 不自增的原因： 因?yàn)槭窍⒁呀?jīng)發(fā)出去了，即使失敗了也不會(huì)重回隊(duì)列，而是直接重新發(fā)一遍消息.
好處： 不僅可以控制重試次數(shù)（防止類似于上面講到的確認(rèn)應(yīng)答引起的無限重試），還可以控制每次重試的間隔時(shí)間（防止負(fù)載飆升）.
到此這篇關(guān)于Rabbit高級(jí)特性 - 消息重試機(jī)制的文章就介紹到這了,更多相關(guān)Rabbit消息重試機(jī)制內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
Java中的CAS(Compare-And-Swap)操作示例詳解
CAS是一種原子操作,用于在多線程環(huán)境中實(shí)現(xiàn)無鎖（lock-free）的線程安全編程,下面通過本文給大家介紹Java中的CAS(Compare-And-Swap)操作示例詳解,感興趣的朋友一起看看吧2025-06-06
Mybatis批量更新對(duì)象數(shù)據(jù)的兩種實(shí)現(xiàn)方式
這篇文章主要介紹了Mybatis批量更新對(duì)象數(shù)據(jù)的兩種實(shí)現(xiàn)方式,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2024-08-08
MybatisPlus自動(dòng)填充時(shí)間的配置類實(shí)現(xiàn)
本文介紹了如何在MyBatis-Plus中實(shí)現(xiàn)自動(dòng)填充時(shí)間的功能,通過實(shí)現(xiàn)MetaObjectHandler接口,重寫insertFill()和updateFill()方法,分別在插入和更新時(shí)填充創(chuàng)建時(shí)間和更新時(shí)間,感興趣的可以了解一下2024-12-12
Java中使用數(shù)組實(shí)現(xiàn)棧數(shù)據(jù)結(jié)構(gòu)實(shí)例
這篇文章主要介紹了Java中使用數(shù)組實(shí)現(xiàn)棧數(shù)據(jù)結(jié)構(gòu)實(shí)例,本文先是講解了實(shí)現(xiàn)棧至少應(yīng)該包括以下幾個(gè)方法等知識(shí),然后給出代碼實(shí)例,需要的朋友可以參考下2015-01-01
SpringBoot+Eureka實(shí)現(xiàn)微服務(wù)負(fù)載均衡的示例代碼
這篇文章主要介紹了SpringBoot+Eureka實(shí)現(xiàn)微服務(wù)負(fù)載均衡的示例代碼，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2019-11-11
在MyBatisPlus中，常常使用@Version實(shí)現(xiàn)樂觀鎖，該注解用于字段上面。本文將通過示例詳細(xì)講解@Version注解的使用，感興趣的可以了解一下2022-06-06
java遍歷http請(qǐng)求request的所有參數(shù)實(shí)現(xiàn)方法
下面小編就為大家?guī)硪黄猨ava遍歷http請(qǐng)求request的所有參數(shù)實(shí)現(xiàn)方法。小編覺得挺不錯(cuò)的，現(xiàn)在就分享給大家，也給大家做個(gè)參考。一起跟隨小編過來看看吧2016-09-09
這篇文章主要介紹了Maven下 mybatis-generator使用 ,非常不錯(cuò)，具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2018-09-09
java 靜態(tài)代理 動(dòng)態(tài)代理深入學(xué)習(xí)
代理模式是常用的java設(shè)計(jì)模式，特征是代理類與委托類有同樣的接口，代理類主要負(fù)責(zé)為委托類預(yù)處理消息、過濾消息、把消息轉(zhuǎn)發(fā)給委托類，以及事后處理消息等,需要的朋友可以參考下2012-11-11
SpringBoot默認(rèn)使用HikariDataSource數(shù)據(jù)源方式
這篇文章主要介紹了SpringBoot默認(rèn)使用HikariDataSource數(shù)據(jù)源方式，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2021-10-10