MQ釋出确認springboot版本

确認機制方案

代碼架構圖  

配置檔案  

spring.rabbitmq.publisher-confirm-type=correlated

⚫ NONE

禁用釋出确認模式，是預設值

⚫ CORRELATED

釋出消息成功到交換器後會觸發回調方法

⚫ SIMPLE

其一效果和CORRELATED值一樣會觸發回調方法， 其二在釋出消息成功後使用rabbitTemplate調用waitForConfirms或waitForConfirmsOrDie方法 等待broker節點傳回發送結果，根據傳回結果來判定下一步的邏輯，要注意的點是 waitForConfirmsOrDie方法如果傳回false則會關閉channel，則接下來無法發送消息到broker

spring.application.name=demo6
spring.rabbitmq.host=192.168.231.135
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123
server.port=8989
spring.rabbitmq.publisher-confirm-type=CORRELATED      

添加配置類 

/**
 * 釋出确認的配置類
 */
@Configuration
@Slf4j
public class config {
    public static  final  String CONFIRM_EXCHANGE_NAME="confirm.exchange";
    public static  final  String CONFIRM_QUEUE_NAME="confirm.queue";
    //聲明業務交換機
    @Bean("confirmExchange")
    public DirectExchange confirmExchange()
    {
        return new DirectExchange(CONFIRM_EXCHANGE_NAME);
    }
    //聲明确認隊列
    @Bean("confirmQueue")
    public Queue confirmQueue()
    {
        return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
    }
    //聲明确認隊列和綁定關系
    @Bean
    public Binding queueBinding(@Qualifier("confirmQueue") Queue queue,@Qualifier("confirmExchange") DirectExchange exchange)
    {
        return BindingBuilder.bind(queue).to(exchange).with("key1");
    }

}      

回調接口  

@Component
@Slf4j
public class MyCallBack implements RabbitTemplate.ConfirmCallback {
    @Override
    public void confirm(CorrelationData correlationData, boolean ack ,String cause) {
        String id=correlationData!=null?correlationData.getId():"";
        if(ack)
        {
            log.info("交換機已經收到id為{}的消息",id);
        }
        else
        {
            log.info("交換機還未收到id未：{}的消息，原因是{}",cause);
        }
    }
}      

消息生産者

@RestController
@RequestMapping("/confirm")
@Slf4j
public class produce {
    public static  final  String confirm_exchange_name="confirm.exchange";
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private  MyCallBack myCallBack;

    @PostConstruct
    public void init()
    {
        rabbitTemplate.setConfirmCallback(myCallBack);
    }
    @GetMapping("sendMessage1/{message}")
    public  void sendMessage1(@PathVariable String message)
    {
        //指定消息id為1
        CorrelationData correlationData1=new CorrelationData("1");
        String routingKey="key1";
        rabbitTemplate.convertAndSend(confirm_exchange_name,routingKey,message+routingKey,correlationData1);
        log.info("發送的消息的内容{}",message);
        CorrelationData correlationData2=new CorrelationData("2");
        routingKey="key2";
        rabbitTemplate.convertAndSend(confirm_exchange_name,routingKey,message+routingKey,correlationData2);
        log.info("發送的消息的内容{}",message);
    }
}      

消息消費者

@Component
@Slf4j
public class ConfirmConsumer {
    public static  final  String CONFIRM_QUEUE_NAME="confirm.queue";
    @RabbitListener(queues = CONFIRM_QUEUE_NAME)
    public void receiveMsg(Message message)
    {
        String s = new String(message.getBody());
        log.info("接收到隊列confirm.queue消息：{}",s);
    }
}      

 發送消息

​​http://localhost:8989/confirm/sendMessage1/3322​​