我要评论消息重试机制
概述
消息重试机制就是在消息处理失败之后重新发送，主要时为了解决消息发送过程可能会出现的问题，例如 网络故障、服务临时不可用 等.
ps：如果时程序逻辑引起的错误，那么即使重试多少次都是没有用的，但是可以通过配置重试次数来解决.
实现方式一：基于消息手动确认机制，返回 nack 实现
配置文件
spring:
application:
name: rabbitmq
rabbitmq:
host: env-base
port: 5672
username: root
password: 1111
listener:
simple:
acknowledge-mode: manual # 手动确认交换机、队列、绑定
@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)
}生产者接口
@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"
}
}消费者
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，并设置 requeue 为 ture 实现消息重新入队，并进行重试
channel.basicnack(deliverytag, false, true)
}
}
}演示和结论
delivertag 自增的原因： 引发异常后，会返回 nack，并且参数 requeue = true，表示重新入队，然后进行重试，将队列中的消息再次发送给生产者，因此 delivertag 会自增.
缺点： 如果是由于程序逻辑异常引起的重试，那么无论重试多少次都没用，并且不断重试会导致负载飙升，性能下降.
实现方式二：基于重试配置实现 配置文件
spring:
application:
name: rabbitmq
rabbitmq:
host: env-base
port: 5672
username: root
password: 1111
listener:
simple:
acknowledge-mode: auto # 开启重试机制，这里必须是 auto，否则不生效！
retry:
enabled: true # 开启消费者失败重试
initial-interval: 5000ms # 失败等待时常
max-attempts: 5 # 最大重试次数(包括第一次消费)ps：开启重试机制，acknowledge-mode 必须指定为 auto，否则不生效！
交换机、队列、绑定
@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)
}生产者接口
@requestmapping("/ack")
fun ack(): string {
rabbittemplate.convertandsend(mqconst.ack_exchange, mqconst.ack_binding, "ack msg 1")
return "ok"
}消费者
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
}
}演示和结论
delivertag 不自增的原因： 因为是消息已经发出去了，即使失败了也不会重回队列，而是直接重新发一遍消息.
好处： 不仅可以控制重试次数（防止类似于上面讲到的确认应答引起的无限重试），还可以控制每次重试的间隔时间（防止负载飙升）.
到此这篇关于rabbit高级特性 - 消息重试机制的文章就介绍到这了,更多相关rabbit消息重试机制内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！
发表评论