Rabbit消息重试机制问题记录

更新时间：2024年08月09日 10:21:25 作者：陈亦康

消息重试机制就是在消息处理失败之后重新发送,主要时为了解决消息发送过程可能会出现的问题,例如网络故障、服务临时不可用等,这篇文章主要介绍了Rabbit消息重试机制问题记录,需要的朋友可以参考下

消息重试机制

概述

消息重试机制就是在消息处理失败之后重新发送，主要时为了解决消息发送过程可能会出现的问题，例如网络故障、服务临时不可用等.

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消息重试机制内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

Java中的CAS(Compare-And-Swap)操作示例详解
CAS是一种原子操作,用于在多线程环境中实现无锁（lock-free）的线程安全编程,下面通过本文给大家介绍Java中的CAS(Compare-And-Swap)操作示例详解,感兴趣的朋友一起看看吧
2025-06-06
Mybatis批量更新对象数据的两种实现方式
这篇文章主要介绍了Mybatis批量更新对象数据的两种实现方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
2024-08-08
MybatisPlus自动填充时间的配置类实现
本文介绍了如何在MyBatis-Plus中实现自动填充时间的功能,通过实现MetaObjectHandler接口,重写insertFill()和updateFill()方法,分别在插入和更新时填充创建时间和更新时间,感兴趣的可以了解一下
2024-12-12
Java中使用数组实现栈数据结构实例
这篇文章主要介绍了Java中使用数组实现栈数据结构实例,本文先是讲解了实现栈至少应该包括以下几个方法等知识,然后给出代码实例,需要的朋友可以参考下
2015-01-01
SpringBoot+Eureka实现微服务负载均衡的示例代码
这篇文章主要介绍了SpringBoot+Eureka实现微服务负载均衡的示例代码，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2019-11-11
详解MybatisPlus中@Version注解的使用
在MyBatisPlus中，常常使用@Version实现乐观锁，该注解用于字段上面。本文将通过示例详细讲解@Version注解的使用，感兴趣的可以了解一下
2022-06-06
java遍历http请求request的所有参数实现方法
下面小编就为大家带来一篇java遍历http请求request的所有参数实现方法。小编觉得挺不错的，现在就分享给大家，也给大家做个参考。一起跟随小编过来看看吧
2016-09-09
Maven下 mybatis-generator使用
这篇文章主要介绍了Maven下 mybatis-generator使用 ,非常不错，具有一定的参考借鉴价值，需要的朋友可以参考下
2018-09-09
java 静态代理动态代理深入学习
代理模式是常用的java设计模式，特征是代理类与委托类有同样的接口，代理类主要负责为委托类预处理消息、过滤消息、把消息转发给委托类，以及事后处理消息等,需要的朋友可以参考下
2012-11-11
SpringBoot默认使用HikariDataSource数据源方式
这篇文章主要介绍了SpringBoot默认使用HikariDataSource数据源方式，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2021-10-10