SpringBoot整合RabbitMq实现ACK机制--消息回退机制--消息确认机制

1. 环境配置

pom.xml

	<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.0.1.RELEASE</version>
    </parent>

    <dependencies>
        <!-- rabbitMQ -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <!-- 测试类 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

application.yml

server:
  port: 9001
spring:
  rabbitmq:
    host: 127.0.0.1
    virtual-host: /
    username: guest
    password: guest
    listener:
      simple:
        prefetch: 1 # 每次只能获取一条消息 处理完成才能获取下一条 能者多劳模式 默认为预取机制
        acknowledge-mode: manual # 设置消费端手动ack确认
        retry:
          enabled: true # 是否支持重试
    publisher-confirm-type: correlated # 确认消息已发送到交换机(Exchange)
    publisher-returns: true #  确认消息已发送到队列(Queue) 消息在未被队列接收时返回

2. RabbitMq配置

RabbitAdmin配置

import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitAdminConfig {

    @Value("${spring.rabbitmq.host}")
    private String host;
    @Value("${spring.rabbitmq.username}")
    private String username;
    @Value("${spring.rabbitmq.password}")
    private String password;
    @Value("${spring.rabbitmq.virtualhost}")
    private String virtualhost;

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setAddresses(host);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualhost);
        // 配置发送确认回调时，次配置必须配置，否则即使在RabbitTemplate配置了ConfirmCallback也不会生效
        connectionFactory.setPublisherConfirms(true);// 开启消息发布者确认
        connectionFactory.setPublisherReturns(true);// 确认消息发送到队列,如未被队列接收时返回
        return connectionFactory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }

2.1 消息发送确认机制

为确保消息发送的准确性，设置发布时确认，确认消息是否到达 Broker 服务器 消息只要被Broker接收,就会触发 ConfirmCallbackConfig 回调

消息接收确认回调

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * 消息只要被RabbitMq Broker接收 就会触发本回调
 * 消息发送确认回调
 */
@Component
public class ConfirmCallbackConfig implements RabbitTemplate.ConfirmCallback {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostConstruct // @PostContruct是spring框架的注解，在⽅法上加该注解会在项⽬启动的时候执⾏该⽅法，也可以理解为在spring容器初始化的时候执行
    public void init() {
        rabbitTemplate.setConfirmCallback(this);
    }

    /**
     * 交换机不管是否收到消息的一个回调方法
     *
     * @param correlationData 消息相关数据
     * @param ack             交换机是否收到消息
     * @param cause           失败原因
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack) { // 消息投递到broker 的状态，true表示成功
            System.out.println("消息发送到Broker成功！");
        } else { // 发送异常
            System.out.println("发送异常原因 = " + cause);
        }
    }
}

2.2 消息投递确认机制

如果消息未能投递到目标 queue 里将触发回调 returnCallback ，一旦向 queue 投递消息未成功，这里一般会记录下当前消息的详细投递数据，方便后续做重发或者补偿等操作

消息投递机制回调接口

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
* 如果消息未能投递到目标queue里将触发本回调 returnCallback ，
 * 一旦向 queue 投递消息未成功，这里一般会记录下当前消息的详细投递数据，方便后续做重发或者补偿等操作
* */
@Component
public class ReturnCallbackConfig implements RabbitTemplate.ReturnCallback {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostConstruct // @PostContruct是spring框架的注解，在⽅法上加该注解会在项⽬启动的时候执⾏该⽅法，也可以理解为在spring容器初始化的时候执
    public void init() {
        rabbitTemplate.setReturnCallback(this);
    }

    @Override
    public void returnedMessage(Message message, int i, String s, String s1, String s2) {
        System.out.println("returnCallback ..............");
        System.out.println(message);// 消息本身
        System.out.println(i);// index
        System.out.println(s);
        System.out.println(s1);
        System.out.println(s2);// 队列名称
    }
}

2.3 ACK消息签收机制

为确保消息 消费成功，需设置消费者消息确认机制，如果消费失败或异常了，可做补偿机制

以下是三种 channel 签收方式

1. basicAck 消息确认
2. basicNack 消息回退
3. basicReject 消息拒绝

消费者消息确认机制

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;


import java.io.IOException;

@Component // 如果把这个注释掉,就不会自动创建 simple.queue队列,则生产消息会触发 returnCallback 回调
public class ACKConfirmListener {
    @RabbitListener(queuesToDeclare = @Queue(value = "simple.queue", durable = "true")) // queuesToDeclare 自动声明队列
    public void ackListener(String msg, Channel channel, Message message) throws IOException {
        try {
            // 消息
            System.out.println("msg = " + msg);
            throw new RuntimeException("消费者故意抛出异常......");
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("消息消费异常，重回队列......");
            /**
             * 出现异常,把消息重新投递回队列中,如一直有异常会一直循环投递
             * deliveryTag：表示消息投递序号。
             * multiple：是否批量确认。
             * requeue：值为 true 消息将重新入队列。
             */
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        }
        /**
         * 消息确认 ACK
         * deliveryTag：表示消息投递序号，每次消费消息或者消息重新投递后，deliveryTag都会增加
         * multiple：是否批量确认，值为 true 则会一次性 ack所有小于当前消息 deliveryTag 的消息。
         */
        System.out.println("ACK消息消费确认.....")
        
        // 消息确认 basicAck
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
       
        // 消息拒绝 basicReject
        //channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);
    }
}

3. 消息生产者

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class MqTest {

    // 注入 rabbitmq
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    // 直接模式（Direct）
    public void testSend() {
        rabbitTemplate.convertAndSend("simple.queue", "ACK消息确认机制生产者......");
    }

https://blog.csdn.net/qq_48721706/article/details/125194646