RabbitMq-消息确认机制-消息队列可靠投递

RabbitMq-消息确认机制-消息队列可靠投递

发送端确认

ConfirmCallback

在spring中开启ConfirmCallback，

springboot rabbitmq属性配置spring.rabbitmq.publisher-confirm和spring.rabbitmq.publisher-confirm-type详解_弃用的配置属性 'spring.rabbitmq.publisher-confirms-CSDN博客

# 这个属性在新版RabbitMq中已经被弃用
# spring.rabbitmq.publisher-confirms = true 
spring.rabbitmq.publisher-confirm-type = correlated #新版本使用这个属性

@Configuration
public class MyRabbitConfig {
    @Autowired
    RabbitTemplate rabbitTemplate;
    
    @PostConstruct //MyRabbitConfig对象创建完成以后，执行这个方法
    public void initRabbitTemplate() {
        // 设置确认回调
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {

            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                System.out.println("...");
            }
        });
    }
}

correlationData是消息的唯一id，在发送消息的时候可以显性设置这个id。如下：

rabbitTemplate.convertAndSend("hello-java-exchange", "hello.java",reasonEntity,new CorrelationData(UUID.randomUUID().toString()));

如果在配置配置类的过程中出现依赖循环，可以试试在application.properties中加上：

spring.main.allow-circular-references=true

returnCallback

配置文件中添加：

#开启发送端 抵达 队列的确认
spring.rabbitmq.publisher-returns=true

spring.rabbitmq.template.mandatory=true

在配置文件中添加：

rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returnedMessage) {
                System.out.println(returnedMessage.toString());
            }
        });

在更新的spring版本中，使用ReturnsCallback接口进行回调。

完整的配置文件如下：ConfirmCallback和returnCallback，

@Configuration
public class MyRabbitConfig {
    @Autowired
    RabbitTemplate rabbitTemplate;

    @PostConstruct //确保rabbitTemplate注入完成后，执行这个方法，否则可能会返回空指针异常
    public void initRabbitTemplate() {
        // 设置确认回调
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {

            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                System.out.println("回调成功" + b);
            }
        });

        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returnedMessage) {
                System.out.println(returnedMessage.toString());
            }
        });
    }
}

ack

配置文件中添加：

spring.rabbitmq.listener.simple.acknowledge-mode=manual

默认确认模式为Auto（自动确认），其余取值有Manual（手动确认）和None（不确认）。

自动确认模式下，当方法获取到消息就立即向mq服务器返回确认，而不是等待直到方法完成。

配置手动确认：

long deliveryTag=message.getMessageProperties().getDeliveryTag();
try{
	// false表示仅确认当前的消息
	channel.basicAck(deliveryTag,false);
}catch(Exception e){
	//网路中断
}

deliveryTag是 RabbitMQ 为每条消息生成的唯一标识符，用于消息的确认和重试机制。它是一个单调递增的 64 位长整型值，每个通道（Channel）独立维护，因此在每个通道上是唯一的。

举例：一条通道内的第一条消息，它的deliveryTag是1，第二条则是2，第三条是3，如此递增…

不确认和拒签

try{
	//不确认
	// 第一个false的参数名为multiple，表示仅是否批量处理；第二个false的参数名为requeue，表示仅是否重新入队
    //long deliveryTag, boolean mutiple, boolean requeue
	channel.basicNack(deliveryTag,false,false);
	
	//拒签
	//long deliveryTag, boolean requeue
	channel.basicReject(deliveryTag,false);
}catch(Exception e){
	//网路中断
}