SpringBoot-MQ代码实现

pom依赖与yml配置

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

spring:
  rabbitmq:
    addresses: 172.28.0.132:5672
    username: admin
    password: aompMq123!
    virtual-host: local-zhengpei

1.直连交换机
使用下direct exchange(直连型交换机),创建DirectRabbitMqConfig.java（对于队列和交换机持久化以及连接使用设置，在注释里有说明，后面的不同交换机的配置就不做同样说明了）：

package com.rabbitmq.demo.config;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
 * Direct交换机 配置
 * @author beyond
 * @since 2020/11/28
 */
@Configuration
public class DirectRabbitMqConfig {
  /**
   * 队列名称 起名 testDirectQueue
   * @return
   */
  @Bean
  public Queue testDirectQueue(){
    // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
    // exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
    // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
    //   return new Queue("testDirectQueue",true,true,false);
    //一般设置一下队列的持久化就好,其余两个就是默认false
    return new Queue("testDirectQueue",true);
  }
  /**
   * Direct交换机 起名：directExchange
   * @return
   */
  @Bean
  DirectExchange directExchange(){
    //  return new DirectExchange("directExchange",true,true);
    return new DirectExchange("directExchange");
  }
  /**
   *   绑定  将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
   */
  @Bean
  Binding bindingDirect() {
    return BindingBuilder.bind(testDirectQueue()).
        to(directExchange()).with("TestDirectRouting");
  }
  @Bean
  DirectExchange lonelyDirectExchange() {
    return new DirectExchange("lonelyDirectExchange");
  }
   /**
    * 将消息转换为json格式,非class格式
    */
  @Bean
  MessageConverter messageConverter() {
    Jackson2JsonMessageConverter jsonMessageConverter = new Jackson2JsonMessageConverter();
    jsonMessageConverter.setDefaultCharset(StandardCharsets.UTF_8.name());
    return jsonMessageConverter;
  }
}

简单的接口进行消息推送,

@RestController
public class SendMessageController {
  @Autowired
  RabbitTemplate rabbitTemplate;  //使用RabbitTemplate,这提供了接收/发送等等方法
  String CREATE_TIME = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
  String MESSAGE_ID = String.valueOf(UUID.randomUUID());
  @GetMapping("/send")
  public String sendMessage(){
    String messageId = MESSAGE_ID;
    String messageData = "test message, hello!";
    String createTime = CREATE_TIME;
    Map<String,Object> map=new HashMap<>();
    map.put("messageId",messageId);
    map.put("messageData",messageData);
    map.put("createTime",createTime);
    //将消息携带绑定键值：TestDirectRouting 发送到交换机directExchange
    rabbitTemplate.convertAndSend("directExchange", "TestDirectRouting", map);
    return messageId;
  }
}

然后是创建消息接收监听类，DirectReceiver.java：

@Component
public class DirectReceiver {
  private static final Logger logger = LoggerFactory.getLogger(DirectReceiver.class);
  // @RabbitHandler 可加可不加
  @RabbitListener(queues = "testDirectQueue")//监听的队列名称 testDirectQueue  
  public void process(Map testMessage) {
    logger.info("DirectReceiver消费者收到消息:[{}]",testMessage.toString());
  }
}

那么直连交换机既然是一对一，那如果咱们配置多台监听绑定到同一个直连交互的同一个队列，会怎么样？

可以看到是实现了轮询的方式对消息进行消费，而且不存在重复消费。

2.**Topic Exchange 主题交换机**
首先创建TopicRabbitConfig.java类，交换机并绑定队列

/**
 * topic 主题交换机配置
 * @author beyond
 * @since 2020/11/29
 */
@Configuration
public class TopicRabbitConfig {
  //绑定键
  public final static String SINGLE = "topic.single";
  public final static String TOTAL = "topic.total";
  @Bean
  Queue firstQueue(){
    return new Queue(SINGLE);
  }
  @Bean
  Queue secondQueue(){
    return new Queue(TOTAL);
  }
  @Bean
  TopicExchange topicExchange(){
    return new TopicExchange("topic.exchange");
  }
  //将firstQueue和topicExchange绑定,而且绑定的键值为topic.single
  //这样只要是消息携带的路由键是topic.single,才会分发到该队列
  @Bean
  Binding bindingExchangeMessage(){
    return BindingBuilder.bind(firstQueue()).to(topicExchange()).with(SINGLE);
  }
  //将secondQueue和topicExchange绑定,而且绑定的键值为用上通配路由键规则topic.#
  // 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列
  @Bean
  Binding bindingExchangeMessage2() {
    return BindingBuilder.bind(secondQueue()).to(topicExchange()).with("topic.#");
  }
}

然后添加多2个接口，用于推送消息到主题交换机：

 @GetMapping("/single")
  public String sendTopicMessage(){
    String messageId = MESSAGE_ID;
    String messageData = "message: single ";
    String createTime = CREATE_TIME;
    Map<String,Object> map=new HashMap<>();
    map.put("messageId",messageId);
    map.put("messageData",messageData);
    map.put("createTime",createTime);
    rabbitTemplate.convertAndSend("topic.exchange", TopicRabbitConfig.SINGLE, map);
    return messageId;
  }
  @GetMapping("/total")
  public String sendTopicMessage2(){
    String messageId = MESSAGE_ID;
    String messageData = "message: topic.total is all ";
    String createTime = CREATE_TIME;
    Map<String,Object> map=new HashMap<>();
    map.put("messageId",messageId);
    map.put("messageData",messageData);
    map.put("createTime",createTime);
    rabbitTemplate.convertAndSend("topic.exchange", TopicRabbitConfig.TOTAL, map);
    return messageId;
  }

生产者这边已经完成，继续创建TopicSingleReceiver.java：

@Component
public class TopicSingleReceiver {
  private static final Logger logger = LoggerFactory.getLogger(TopicSingleReceiver.class);
  @RabbitHandler
  @RabbitListener(queues = TopicRabbitConfig.SINGLE) //监听的队列名称 topic.single
  public void process(Map testMessage) {
    logger.info("TopicSingleReceiver消费者收到消息:[{}]",testMessage.toString());
  }
}

再创建一个TopicTotalReceiver.java：

@Component
@RabbitListener(queues = TopicRabbitConfig.TOTAL) //监听的队列名称 topic.total
public class TopicTotalReceiver {
  private static final Logger logger = LoggerFactory.getLogger(TopicTotalReceiver.class);
  @RabbitHandler
  public void process(Map testMessage) {
    logger.info("TopicTotalReceiver消费者收到消息:[{}]",testMessage.toString());
  }
}

查看topic交换机绑定情况

先调用/single接口

然后看消费者的控制台输出情况：
TopicSingleReceiver监听队列1，绑定键为：topic.single
TopicTotalReceiver监听队列2，绑定键为：topic.#
而当前推送的消息，携带的路由键为：topic.single
所以可以看到两个监听消费者receiver都成功消费到了消息，因为这两个recevier监听的队列的绑定键都能与这条消息携带的路由键匹配上。

接下来调用接口/total:

可以看到两个监听消费者只有TopicTotalReceiver成功消费到了消息。

3.**Fanout Exchang 扇型交换机**
创建类FanoutRabbitConfig.java

/**
 * 扇形交换机配置
 * @author beyond
 * @since 2020/11/29
 */
@Configuration
public class FanoutRabbitConfig {
  /**
   *  创建三个队列 ：fanout.first   fanout.second  fanout.three
   *  将三个队列都绑定在交换机 fanout.exchange 上
   *  因为是扇型交换机, 路由键无需配置,配置也不起作用
   */
  public static final String FIRST = "fanout.first";
  public static final String SECOND = "fanout.second";
  public static final String THREE = "fanout.three";
  @Bean
  Queue first(){
    return new Queue(FIRST);
  }
  @Bean
  Queue second(){
    return new Queue(SECOND);
  }
  @Bean
  Queue three(){
    return new Queue(THREE);
  }
  @Bean
  FanoutExchange fanoutExchange(){
    return new FanoutExchange("fanout.exchange");
  }
  @Bean
  Binding bindingExchangeFirst(){
    return BindingBuilder.bind(first()).to(fanoutExchange());
  }
  @Bean
  Binding bindingExchangeSecond(){
    return BindingBuilder.bind(second()).to(fanoutExchange());
  }
  @Bean
  Binding bindingExchangeThree(){
    return BindingBuilder.bind(three()).to(fanoutExchange());
  }
}

写一个接口用于推送消息,

  @GetMapping("/fanout")
  public Map<String, Object> sendFanoutMessage() {
    String messageId = MESSAGE_ID;
    String messageData = "message: testFanoutMessage ";
    String createTime = CREATE_TIME;
    Map<String, Object> map = new HashMap<>();
    map.put("messageId", messageId);
    map.put("messageData", messageData);
    map.put("createTime", createTime);
    rabbitTemplate.convertAndSend("fanout.exchange", null, map);
    return map;
  }

创建三个消息消费类进行消费，FanoutReceiverFirst.java, FanoutReceiverSecond.java,FanoutReceiverThree.java分别如下

@Component
public class FanoutReceiverFirst {
  private static final Logger logger = LoggerFactory.getLogger(FanoutReceiverFirst.class);
  @RabbitHandler
  @RabbitListener(queues = "fanout.first")
  public void process(Map testMessage) {
    logger.info("FanoutReceiverFirst消费者收到消息:[{}]",testMessage.toString());
  }
}

@Component
public class FanoutReceiverSecond {
  private static final Logger logger = LoggerFactory.getLogger(FanoutReceiverSecond.class);
  @RabbitHandler
  @RabbitListener(queues = "fanout.second")
  public void process(Map testMessage) {
    logger.info("FanoutReceiverSecond消费者收到消息:[{}]",testMessage.toString());
  }
}

@Component
public class FanoutReceiverThree {
  private static final Logger logger = LoggerFactory.getLogger(FanoutReceiverThree.class);
  @RabbitHandler
  @RabbitListener(queues = "fanout.three")
  public void process(Map testMessage) {
    logger.info("FanoutReceiverThree消费者收到消息:[{}]",testMessage.toString());
  }
}

调用接口/fanout，查看控制台输出
可以看到只要发送到 fanoutExchange 这个扇型交换机的消息， 三个队列都绑定这个交换机，所以三个消息接收类都监听到了这条消息。

到了这里其实三个常用的交换机的使用我们已经完毕了，那么接下来我们继续讲讲消息的回调，其实就是消息确认（生产者推送消息成功，消费者接收消息成功）。

如果在配置确认回调，测试发现无法触发回调函数，那么存在原因也许是因为版本导致的配置项不起效，
可以把publisher-confirm-type: correlated 替换为 publisher-confirms: true

spring:
  rabbitmq:
    addresses: 172.28.0.132:5672
    username: admin
    password: aompMq123!
    virtual-host: local-zhengpei
      #确认消息已发送到交换机(Exchange)
    publisher-confirm-type: correlated
      #确认消息已发送到队列(Queue)
    publisher-returns: true

配置相关的消息确认回调函数，RabbitConfig.java：

package com.rabbitmq.demo.config;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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;
/**
 * 消息确认
 *
 * @author beyond
 * @since 2020/11/29
 */
@Configuration
public class RabbitMqConfig {
  private static final Logger logger = LoggerFactory.getLogger(RabbitMqConfig.class);
  @Bean
  public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
    RabbitTemplate template = new RabbitTemplate();
    template.setConnectionFactory(connectionFactory);
    //设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数
    template.setMandatory(true);
    template.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
      @Override
      public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        logger.info("ConfirmCallback:相关数据:[{}]", correlationData);
        logger.info("ConfirmCallback:确认情况:[{}]", ack);
        logger.info("ConfirmCallback:原因:[{}]", cause);
      }
    });
    template.setReturnCallback(new RabbitTemplate.ReturnCallback() {
      @Override
      public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        logger.info("ReturnCallback消息：[{}]", message);
        logger.info("ReturnCallback回应码：[{}]", replyCode);
        logger.info("ReturnCallback回应信息：[{}]", replyText);
        logger.info("ReturnCallback交换机：[{}]", exchange);
        logger.info("ReturnCallback路由键：[{}]", routingKey);
      }
    });
    return template;
  }
}

到这里，生产者推送消息的消息确认调用回调函数已经完毕。
可以看到上面写了两个回调函数，一个叫 ConfirmCallback ，一个叫 RetrunCallback；
那么以上这两种回调函数都是在什么情况会触发呢？
先从总体的情况分析，推送消息存在四种情况：
①消息推送到server，但是在server里找不到交换机
②消息推送到server，找到交换机了，但是没找到队列
③消息推送到sever，交换机和队列啥都没找到
④消息推送成功
那么先写几个接口来分别测试和认证下以上4种情况，消息确认触发回调函数的情况
①消息推送到server，但是在server里找不到交换机
写个测试接口，把消息推送到名为‘non-existent-exchange’的交换机上（这个交换机是没有创建没有配置的）：

1. @GetMapping("/test-message-ack")
  public Map<String, Object> TestMessageAck() {
    String messageId = MESSAGE_ID;
    String messageData = "message: non-existent-exchange test message ";
    String createTime = CREATE_TIME;;
    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 map;
  }

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

 Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'non-existent-exchange' in vhost 'local-zhengpei', class-id=60, method-id=40)
 ConfirmCallback:相关数据:[null]
 ConfirmCallback:确认情况:[false]
 ConfirmCallback:原因:[channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'non-existent-exchange' in vhost 'local-zhengpei', class-id=60, method-id=40)]

结论： ①这种情况触发的是 ConfirmCallback 回调函数。<br />**②消息推送到server，找到交换机了，但是没找到队列  **<br />这种情况就是需要新增一个交换机，但是不给这个交换机绑定队列，我来简单地在DirectRabitConfig里面新增一个直连交换机，名叫‘lonelyDirectExchange’，但没给它做任何绑定配置操作

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

调用结果查看情况

  @GetMapping("/test-message-ack2")
  public Map<String, Object> TestMessageAck2() {
    String messageData = "message: lonelyDirectExchange test message ";
    Map<String, Object> map = new HashMap<>();
    map.put("messageId", MESSAGE_ID);
    map.put("messageData", messageData);
    map.put("createTime", CREATE_TIME);
    rabbitTemplate.convertAndSend("lonelyDirectExchange", "TestDirectRouting", map);
    return map;
  }

ConfirmCallback:相关数据:[null]
ConfirmCallback:确认情况:[true]
ConfirmCallback:原因:[null]
ReturnCallback消息：[(Body:'{createTime=2020-11-29 20:28:55, messageId=0a916844-63cf-463e-a681-3be92b148ace, messageData=message: lonelyDirectExchange test message }' MessageProperties [headers={}, contentType=application/x-java-serialized-object, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, deliveryTag=0])]
ReturnCallback回应码：[312]
ReturnCallback回应信息：[NO_ROUTE]
ReturnCallback交换机：[lonelyDirectExchange]
ReturnCallback路由键：[TestDirectRouting]

可以看到这种情况，两个函数都被调用了；
这种情况下，消息是推送成功到服务器了的，所以ConfirmCallback对消息确认情况是true；
而在RetrunCallback回调函数的打印参数里面可以看到，消息是推送到了交换机成功了，但是在路由分发给队列的时候，找不到队列，所以报了错误 NO_ROUTE 。
结论：②这种情况触发的是 ConfirmCallback和RetrunCallback两个回调函数。

③消息推送到sever，交换机和队列啥都没找到
这种情况其实一看就觉得跟①很像，没错 ，③和①情况回调是一致的，所以不做结果说明了。
结论： ③这种情况触发的是 ConfirmCallback 回调函数。

④消息推送成功
那么测试下，按照正常调用之前消息推送的接口就行，就调用下 /sendFanoutMessage接口，可以看到控制台输出：

ConfirmCallback:     相关数据：null
ConfirmCallback:     确认情况：true
ConfirmCallback:     原因：null

结论： ④这种情况触发的是 ConfirmCallback 回调函数。
以上是生产者推送消息的消息确认 回调函数的使用介绍（可以在回调函数根据需求做对应的扩展或者业务数据处理）。
.

==========================消息确认机制=====================================

和生产者的消息确认机制不同，因为消息接收本来就是在监听消息，符合条件的消息就会消费下来。
所以，消息接收的确认机制主要存在三种模式：
①自动确认， 这也是默认的消息确认情况。 AcknowledgeMode.NONE
RabbitMQ成功将消息发出（即将消息成功写入TCP Socket）中立即认为本次投递已经被正确处理，不管消费者端是否成功处理本次投递。
所以这种情况如果消费端消费逻辑抛出异常，也就是消费端没有处理成功这条消息，那么就相当于丢失了消息。
一般这种情况我们都是使用try catch捕捉异常后，打印日志用于追踪数据，这样找出对应数据再做后续处理。
② 根据情况确认， 这个不做介绍
③ 手动确认 ， 这个比较关键，也是我们配置接收消息确认机制时，多数选择的模式。
消费者收到消息后，手动调用basic.ack/basic.nack/basic.reject后，RabbitMQ收到这些消息后，才认为本次投递成功。
basic.ack用于肯定确认
basic.nack用于否定确认（注意：这是AMQP 0-9-1的RabbitMQ扩展）
basic.reject用于否定确认，但与basic.nack相比有一个限制:一次只能拒绝单条消息

消费者端以上的3个方法都表示消息已经被正确投递，但是basic.ack表示消息已经被正确处理。
而basic.nack,basic.reject表示没有被正确处理：
着重讲下reject，因为有时候一些场景是需要重新入列的。
channel.basicReject(deliveryTag, true); 拒绝消费当前消息，如果第二参数传入true，就是将数据重新丢回队列里，那么下次还会消费这消息。设置false，就是告诉服务器，我已经知道这条消息数据了，因为一些原因拒绝它，而且服务器也把这个消息丢掉就行。 下次不想再消费这条消息了。
使用拒绝后重新入列这个确认模式要谨慎，因为一般都是出现异常的时候，catch异常再拒绝入列，选择是否重入列。
但是如果使用不当会导致一些每次都被你重入列的消息一直消费-入列-消费-入列这样循环，会导致消息积压。

顺便也简单讲讲 nack，这个也是相当于设置不消费某条消息。
channel.basicNack(deliveryTag, false, true);
第一个参数依然是当前消息到的数据的唯一id;
第二个参数是指是否针对多条消息；如果是true，也就是说一次性针对当前通道的消息的tagID小于当前这条消息的，都拒绝确认。
第三个参数是指是否重新入列，也就是指不确认的消息是否重新丢回到队列里面去。
同样使用不确认后重新入列这个确认模式要谨慎，因为这里也可能因为考虑不周出现消息一直被重新丢回去的情况，导致积压。

实现手动确认 ？？？
**