13.RabbitMQ 消息可靠性投递confirm确认模式

Posted 2021-08-20 九宫格输入法

  在使用RabbitMQ的时候，消息生产者发送消息不希望出现消息丢失或者投递失败的现象；

  RabbitMQ在消息投递可靠性方面给我们提供了两种模式；

    1，confirm确认模式

    2，return 退回模式

confirm确认模式介绍以及实现

  confirm机制，消息的确认，是指生产者投递消息之后，如果Broker收到消息，则会给生产者一个应答，生产者能接收应答，用来确定这条消息是否正常的发送到Broker，这种机制是消息可靠性投递的核心保障。

  confirm机制是只保证消息到达exchange，并不保证消息可以路由到正确的queue，如果exchange错误，就会触发confirm机制
  1、配置文件设置

# 开启confirm模式
spring.rabbitmq.publisher-confirm-type=correlated

  2、com.java996.producer.Service.impl.RabbitMqServiceImpl实现RabbitTemplate.ConfirmCallback接口，实现confirm方法

@Service("RabbitMqService")
public class RabbitMqServiceImpl implements RabbitMqService, RabbitTemplate.ConfirmCallback{

    /**
     *
     * @param correlationData  消息唯一标识
     * @param ack 交换机是否成功收到消息 true成功  false失败
     * @param cause 失败原因
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        System.out.println("confirm方法被执行..."+correlationData.getId());
        if(ack){
            System.out.println("交换机 消息接收成功"+cause);
        }else{
            System.out.println("交换机 消息接收失败"+cause);
            // 这里我们需要做一些消息补发措施
        }
    }
    ......    
}

  3、注入RabbitTemplate模版

@Autowired
private RabbitTemplate rabbitTemplate;

  4、初始化设置

@PostConstruct
public void init(){
    rabbitTemplate.setConfirmCallback(this);
}

  5、发送数据测试

@Override
public void sendMessage(String message) {
    CorrelationData correlationData = new CorrelationData("1");
    rabbitTemplate.convertAndSend(RabbitMqConfig.DIRECT_EXCHANGE,RabbitMqConfig.DIRECT_ROUTINGKEY,message,correlationData);
}

  6、启动测试:

package com.java996.producer;

import com.java996.producer.Service.RabbitMqService;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;

@SpringBootApplication
public class ProducerApp {
    public static void main(String[] args) {
     ConfigurableApplicationContext context = SpringApplication.run(ProducerApp.class,args);
        RabbitMqService rabbitMqService = (RabbitMqService) context.getBean("RabbitMqService");
        rabbitMqService.sendMessage("confirm确认测试");
    }
}

  7、控制台：

  8、队列：