SpringBoot整合RabbitMQ实现消息确认机制_Java教程

前面几篇案例已经将常用的交换器（DirectExchange、TopicExchange、FanoutExchange）的用法介绍完了，现在我们来看一下消息的回调，也就是消息确认。

在rabbitmq-provider项目的application.yml文件上加上一些配置

server:
port: 8021
spring:
#给项目来个名字
application:
  name: rabbitmq-provider
#配置rabbitMq 服务器
rabbitmq:
  host: 127.0.0.1
  port: 5672
  username: guest
  password: guest
#    connection-timeout: 60s
  #虚拟host 可以不设置,使用server默认host
  #virtual-host: admin
  # 确认消息已发送到交换机（Exchange）
  publisher-confirm-type: correlated
  # 确认消息已发送到队列
  publisher-returns: true

创建RabbitConfig配置类，配置消息的相关回调函数

package com.rabbitmq.provider.config;

import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@Configuration
public class RabbitConfig {

  @Bean
  DirectExchange lonelyDirectExchange() {
      return new DirectExchange("lonelyDirectExchange");
  }

  @Bean
  public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory){
      RabbitTemplate rabbitTemplate = new RabbitTemplate();
      rabbitTemplate.setConnectionFactory(connectionFactory);
      // 设置开启Mandatory，才能触发回调函数，无论消息推送结果怎么样都强制调用回调函数
      rabbitTemplate.setMandatory(true);
      rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
          @Override
          public void confirm(CorrelationData correlationData, boolean b, String s) {
              System.out.println("ConfirmCallback    ：相关数据：" + correlationData);
              System.out.println("ConfirmCallback    ：确认情况：" + b);
              System.out.println("ConfirmCallback    ：原因：" + s);
          }
      });

      rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
          @Override
          public void returnedMessage(Message message, int i, String s, String s1, String s2) {
              System.out.println("ReturnCallback    ：消息：" + message);
              System.out.println("ReturnCallback    ：回应码：" + i);
              System.out.println("ReturnCallback    ：回应消息：" + s);
              System.out.println("ReturnCallback    ：交换机：" + s1);
              System.out.println("ReturnCallback    ：路由键：" + s2);
          }
      });
      return rabbitTemplate;
  }
}

生产者回调函数已经配置完毕，上面我们配置了两个回调函数setConfirmCallback和setReturnCallback

一般会产生四种情况：

①消息推送到server，但是在server里找不到交换机
②消息推送到server，找到交换机了，但是没找到队列
③消息推送到sever，交换机和队列啥都没找到
④消息推送成功

接下来分别测试和认证下以上4种情况，消息确认触发回调函数的情况：

①消息推送到server，但是在server里找不到交换机

写个测试接口，把消息推送到名为‘non-existent-exchange'的交换机上（这个交换机是没有创建没有配置的）：

@GetMapping("/testMessageAck")
  public String testMessageAck() {
      String messageId = String.valueOf(UUID.randomUUID());
      String messageData = "message: non-existent-exchange test message ";
      String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
      Map<String, Object> map = new HashMap<>();
      map.put("messageId", messageId);
      map.put("messageData", messageData);
      map.put("createTime", createTime);
      rabbitTemplate.convertAndSend("non-existent-exchange", "TestDirectRouting", map);
      return "ok";
  }

调用接口，查看rabbitmq-provuder项目的控制台输出情况（原因里面有说，没有找到交换机'non-existent-exchange'）：

SpringBoot整合RabbitMQ实现消息确认机制

所以证明在交换器没有配置，推送消息找不到对应的交换器，会执行ConfirmCallback方法。

②消息推送到server，找到交换机了，但是没找到队列

我们在RabbitConfig里面配置了一个名为lonelyDirectExchange的交换器，再新增一个接口testMessageAck2，使用lonelyDirectExchange交换器

@GetMapping("/testMessageAck2")
  public String testMessageAck2() {
      String messageId = String.valueOf(UUID.randomUUID());
      String messageData = "message: non-existent-exchange test message ";
      String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
      Map<String, Object> map = new HashMap<>();
      map.put("messageId", messageId);
      map.put("messageData", messageData);
      map.put("createTime", createTime);
      rabbitTemplate.convertAndSend("lonelyDirectExchange", "TestDirectRouting", map);
      return "ok";
  }

重启服务，使用postman调用这个接口

SpringBoot整合RabbitMQ实现消息确认机制

可以看到ReturnCallback和ConfirmCallback这两个都被调用了。

CallfirmCallback的确认情况为true，因为消息是推送成功到服务器了的，所以ConfirmCallback对消息确认情况是true

ReturnCallback的回应消息为NO_ROUTE找不到队列

③消息推送到sever，交换机和队列啥都没找到

这种情况和1、2很相似，会调用ReturnCallback和CallfirmCallback这两个回调函数

④消息推送成功

我们调用之前的sendFanoutMessage接口来看一下会是什么情况

SpringBoot整合RabbitMQ实现消息确认机制

结论：消息推送成功调用ConfirmCallback回调函数

消息确认机制

消息接收确认机制主要存在三种模式：

①：自动确认

②：根据情况确认

③：手动确认

接下来，我们看看手动确认是怎么实现的

在rabbitmq-consumer消费者项目上创建MessageListenerConfig配置类

package com.rabbitmq.consumer.config;

import com.rabbitmq.consumer.receiver.MyAckReceiver;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@Configuration
public class MessageListenerConfig {

  @Autowired
  private CachingConnectionFactory connectionFactory;

  @Autowired
  private MyAckReceiver myAckReceiver;    //消息接收处理类

  @Bean
  public SimpleMessageListenerContainer simpleMessageListenerContainer() {
      SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
      container.setConcurrentConsumers(1);
      container.setMaxConcurrentConsumers(1);
      // RabbitMQ 默认是自动确认，改成手动确认
      container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
      // 设置队列
      // 如果同时设置多个如下:前提是队列是必须已经创建存在的
      // container.setQueueNames("TestDirectQueue","TestDirectQueue2","TestDirectQueue3");
      // 设置一个队列
      // container.setQueues(new Queue("TestDirectQueue",true));
      // 设置多个队列
       container.addQueues(new Queue("TestDirectQueue",true));
       container.addQueues(new Queue("fanout.A",true));
      // container.addQueues(new Queue("TestDirectQueue3",true));

      container.setMessageListener(myAckReceiver);
      return container;
  }

}

创建手动确定消息监听类MyAckReceiver

package com.rabbitmq.consumer.receiver;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;


@Component
public class MyAckReceiver implements ChannelAwareMessageListener {


  @Override
  public void onMessage(Message message, Channel channel) throws Exception {
      long deliveryTag = message.getMessageProperties().getDeliveryTag();
      try {
          //因为传递消息的时候用的map传递，所以将map从Message内取出来需要做些处理
          String msg = message.toString();
          String[] msgArray = msg.split("'");
          Map<String,String> msgMap = mapStringToMap(msgArray[1].trim(),3);
          String messageId = msgMap.get("messageId");
          String messageData = msgMap.get("messageData");
          String createTime = msgMap.get("createTime");
          System.out.println("MyAckReceiver ：");
          System.out.println("messageId = " + messageId);
          System.out.println("messageData = " + messageData);
          System.out.println("createTime = " + createTime);
          System.out.println("消费的主题来自：" + message.getMessageProperties().getConsumerQueue());
          if("TestDirectQueue".equals(message.getMessageProperties().getConsumerQueue())){
              System.out.println("执行TestDirectQueue中的消息的业务处理流程.....");
          }
          if("fanout.A".equals(message.getMessageProperties().getConsumerQueue())){
              System.out.println("执行fanout.A中的消息的业务处理流程.....");
          }
          //第二个参数，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息
          channel.basicAck(deliveryTag,true);
          //第二个参数，true会重新放回队列，所以需要自己根据业务逻辑判断什么时候使用拒绝
          //channel.basicReject(deliveryTag, true);
      }catch (Exception e){
          channel.basicAck(deliveryTag,false);
          e.printStackTrace();
      }
  }
  // {key=value,key=value,key=value} 格式转换成map
  private Map<String,String> mapStringToMap(String str,int entryNum) {
      str = str.substring(1, str.length()-1);
      String[] strs = str.split(",", entryNum);
      Map<String,String> map = new HashMap<String,String>();
      for (String s : strs) {
          String key = s.split("=")[0].trim();
          String value = s.split("=")[1];
          map.put(key,value);
      }
      return map;
  }
}

重启服务，使用postman调用sendDirectMessage接口

SpringBoot整合RabbitMQ实现消息确认机制