rabbitMQ-生产者可靠发送-消息确认-Confirm_publisher-confirms: true

1、rabbitMQ可能存在的数据不一致问题原因

• 生产者发送消息到Broker消息服务器（消息发送失败）
• rabbitmq服务器自身故障导致消息丢失
• 消息消费者接收消息后处理失败（消费消息失败）

2、生产者发送消息防止消息丢失

•  使用rabbitMq的事务机制，效率极低，并且失去了异步的初衷，所以建议不适用
• 使用confirm消息确认机制，比上面的事务机制要好一些，但是当访问量上来之后，由于频繁的确认交互，也会很大程度降低效率，所以如果不是很重要的消息，也不建议使用。
• 其实confirm机制只适用于生产者发送到exchange交换机的回调，如果交换机没有路由到queue中，就需要开启return机制了。

3、springBoot演示消息确认机制

3.1引入依赖

    <!--引入rabbitmq继承依赖-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
 
 
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit-test</artifactId>
            <scope>test</scope>
        </dependency>

3.2配置application.yml

spring:
  application:
    name: rabbitmq-springboot
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: ems
    password: 123
    virtual-host: /ems
    # 发送者开启 confirm 确认机制
    publisher-confirms: true
    # 发送者开启 return 确认机制
    publisher-returns: true
    listener:
      simple:
        # 设置消费端手动 ack
        acknowledge-mode: manual
        # 是否支持重试
        retry:
          enabled: true

3.3管理交换机、队列的Bean的配置类

package com.lst;
 
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class QueueConfig {
 
    //定义队列
    @Bean(name = "TestQueue")
    public Queue returnTestQueue() {
        return new Queue("returnTestQueue", true, false, false);
    }
 
    //定义交换机
    @Bean(name = "TestExchange")
    public DirectExchange returnTestExchange() {
        return new DirectExchange("TestExchange");
    }
 
    //将队列绑定交换机
    @Bean
    public Binding confirmTestExchangeAndQueue(
            @Qualifier("TestExchange") DirectExchange TestExchange,
            @Qualifier("TestQueue") Queue TestQueue) {
        return BindingBuilder.bind(TestQueue).to(TestExchange).with("testRoutingKey");
    }
}

 3.4生产者测试类

package com.lst;
 
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.connection.CorrelationData;
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;
 
@SpringBootTest(classes = RabbitmqSpringbootApplication.class)
@RunWith(SpringRunner.class)
public class TestRabbitMq {
 
    //注入rabbitTemplate对象
    @Autowired
    private RabbitTemplate rabbitTemplate;
 
    /**
     * 确认模式步骤
     *    1、确认模式开启 publisher-confirms: true
     *    2、在rabbitTemple中定义回调函数
     */
    @Test
    public void test(){
 
        //定义回调
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            /**
             *
             * @param correlationData 相关的配置
             * @param b ：ack  交换机是否成功收到了消息 true代表成功 false代表失败
             * @param s ：失败的原因
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
               if(b){//接收成功
                   System.out.println("发送成功");
               }else{//接受失败
                   System.out.println("接收失败"+s);
               }
            }
        });
        //发送消息
        rabbitTemplate.convertAndSend("TestExchange","testRoutingKey","hello");
 
 
    }
}

3.5 测试结果

进行到这，出现了一个问题，虽然消息成功进入了队列，但是返回的ack一直是false，查阅资料得知，因为我这里的生产者使用test方法模拟，所以当消息发送之后，connection就会被立即销毁，所以。当测试方法结束，rabbitmq相关的资源也就关闭了，虽然我们的消息发送出去，但异步的ConfirmCallback却由于资源关闭而出现了上面的问题。

所以我改用rest风格的接口访问

package com.lst.hello;
 
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;
 
@RestController
public class TestRabbitMq {
 
    //注入rabbitTemplate对象
    @Autowired
    private RabbitTemplate rabbitTemplate;
 
    /**
     * 确认模式步骤
     * 1、确认模式开启 connetionfactory中开启publisher—confirm = "true"
     * 2、在rabbitTemple中定义回调函数
     */
    @PostMapping("/test")
    public void test() {
 
        //定义回调
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            /**
             *
             * @param correlationData 相关的配置
             * @param b ：ack  交换机是否成功收到了消息 true代表成功 false代表失败
             * @param s ：失败的原因
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                if (b) {//接收成功
                    System.out.println("发送成功");
                } else {//接受失败
                    System.out.println("接收失败" + s);
                }
            }
        });
        //发送消息
        rabbitTemplate.convertAndSend("TestExchange", "testRoutingKey", "hello confirm");
 
 
    }
 
}

调用之后

成功示例：

 失败示例：

我们随意更改以下交换机的名称，使其错误，再调用一下