Spring boot 整合 RabbitMQ 消息队列介绍

Spring boot 整合 RabbitMQ，本示例采用 RabbitTemplate 发送消息，采用 @RabbitListener 接受消息。

一、依赖包和配置项

1.1、在 pom 文件中添加依赖信息

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

1.2、添加配置项(自动装配)

spring.rabbitmq.host=127.0.0.1

spring.rabbitmq.port=5672

spring.rabbitmq.username=guest

spring.rabbitmq.password=guest

二、自定义注入配置 Bean 相关组件

在 Spring boot 整合 RabbitMQ 的项目中，为了方便使用 RabbitMQ 的相关操作组件和跟踪消息在发送过程中的状态，可以在项目中自定义注入和配置 Bean 相关组件。下面设置自定义配置的 Bean 组件放到 RabbitmqConfig 配置类中。

package com.example.mgt.examplemgtservicek8stest.config;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;

/**
 * RabbitMQ 自定义注入配置 Bean 相关组件
 * @since 2020/5/15 13:52
 */
@Configuration
public class RabbitmqConfig {
  //定义日志
  private static final Logger log = LoggerFactory.getLogger(RabbitmqConfig.class);

  //自动装配 RabbitMQ 的链接工厂实例
  @Autowired
  private CachingConnectionFactory connectionFactory;
  //自动装配消息监听器所在的容器工厂配置类实例
  @Autowired
  private SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer;


//下面为单一消费者实例的配置
  @Bean(name="singleListenerContainer")
  public SimpleRabbitListenerContainerFactory listenerContainer(){
  //定义消息监听器所在的容器工厂
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
  //设置容器工厂所用的实例
    factory.setConnectionFactory(connectionFactory);
  //设置消息在传输中的格式，这里采用 JSON 的格式进行传输
    factory.setMessageConverter(new Jackson2JsonMessageConverter());
  //设置并发消费者实例的初始数量为1
    factory.setConcurrentConsumers(1);
  //设置并发消费者实例中最大数量为1
    factory.setMaxConcurrentConsumers(1);
  //设置并发消费者实例中每个实例拉取的消息数量为1个
    factory.setPrefetchCount(1);
    return factory;
  }

//自定义配置 RabbitMQ 发送消息的操作组件 RabbitTemplate
  @Bean
  public RabbitTemplate rabbitTemplate(){
  //设置：发送消息后进行确认
    connectionFactory.setPublisherConfirms(true);
  //设置：发送消息后返回确认信息
    connectionFactory.setPublisherReturns(true);
  //构造发送消息组件实例对象
    RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

    rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
    @Override
      public void confirm(
          CorrelationData correlationData, boolean b, String s) {
        if(b){
          log.info("消息发送成功:correlationData({}),ack({}),cause({})",correlationData,b,s);
        }else{
          System.out.println("消息确认失败");
        }

      }
    });

// 设置消息收到确认
rabbitTemplate.setMandatory(true);
/*    rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
    @Override
      public void returnedMessage(
          Message message, int i, String s, String s1, String s2) {
        log.info("消息发送失败");
      }
    });*/

    return rabbitTemplate;
  }
}

1、当mandatory标志位设置为true时，如果exchange根据自身类型和消息routingKey无法找到一个合适的queue存储消息，那么broker会调用basic.return方法将消息返还给生产者;

2、当mandatory设置为false时，出现上述情况broker会直接将消息丢弃;

三、RabbitMQ 发送、接受消息实战

Spring boot 整合 RabbitMQ 的正式代码编写前，需要创建队列、交换机、路由及绑定操作。以生产者发送一个简单字符串为例。

3.1、在 RabbitmqConfig 类中创建需要对象

最先在 RabbitmqConfig 类中创建队列、交换机、路由和绑定操作，如下：

// 定义读取配置文件的环境变量的实例
  @Autowired
  private Environment env;
//创建队列
  @Bean(name = "basicQueue")
  public Queue basicQueue(){
    return new Queue(env.getProperty("mq.basic.info.queue.name"),true);
  }
//创建交换机：以 DirectExchange 为例
  @Bean
  public DirectExchange basicExchange(){
    return new DirectExchange(env.getProperty("mq.basic.info.exchange.name"),true,false);
  }
//绑定
  @Bean
  public Binding basicBinding(){
    return BindingBuilder.bind(basicQueue()).to(basicExchange()).
        with(env.getProperty("mq.basic.info.routing.key.name"));
  }

补充：上面 env 读取到的变量，都需要在配置文件 application.yml 中设置好。

mq:
  basic:
    info:
      queue:
        name: lyz.queue
      exchange:
        name: lyz.exchange
      routing:
        key:
          name: lyz.key

3.2、生产者发送消息

package com.example.mgt.examplemgtservicek8stest.rabbitmq.publisher;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Strings;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;

/**
 * 生产者
 * @since 2020/5/15 14:33
 */
@Component
public class BasicPublisher {
  private static final Logger log = LoggerFactory.getLogger(BasicPublisher.class);
  //定义 JSON 序列化和反序列化实例
  @Autowired
  private ObjectMapper objectMapper;
  //定义 RabbitMQ 消息操作组件
  @Autowired
  private RabbitTemplate rabbitTemplate;

  @Autowired
  private Environment env;

  public void sendMsg(String message) {
    if(!Strings.isNullOrEmpty(message)){
      try{
	  //定义传输格式为 JSON 字符串
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
	  //指定消息模型中的交换机
        rabbitTemplate.setExchange(env.getProperty("mq.basic.info.exchange.name"));
	  //指定消息模型中的路由
        rabbitTemplate.setRoutingKey(env.getProperty("mq.basic.info.routing.key.name"));
	  //将字符串转化为待发送的消息，即一串二进制的数据流
        Message msg = MessageBuilder.withBody(message.getBytes("utf-8")).build();
	  //转化并发送消息
        rabbitTemplate.convertAndSend(msg);
        log.info("基本消息模型-生产者-发送消息：{}",message);
      } catch (Exception e){
        log.error("基本消息模型-生产者-发送消息异常：{}",message,e.fillInStackTrace());
      }
    }
  }
}

3.3、消费者监听并接受消费消息

package com.example.mgt.examplemgtservicek8stest.rabbitmq.consumer;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

/**
 * 消费者
 * @since 2020/5/15 14:57
 */
@Component
public class BasicConsumer {
  private static final Logger log = LoggerFactory.getLogger(BasicConsumer.class);
  @Autowired
  public ObjectMapper objectMapper;
//监听并接受消费队列中的消息-容器工厂 singleListenerContainer 在 RabbitmqConfig 类中定义
  @RabbitListener(queues="${mq.basic.info.queue.name}",containerFactory = "singleListenerContainer")
//由于消息本质是一串二进制数据流，因而监听接受的消息采用字节数组接收
  public void consumeMsg1(@Payload byte[] msg) {
    try{
      String message = new String(msg,"utf-8");
      log.info("基本信息模型-消费者-监听到的消息： {}",message);
    } catch (Exception e){
      log.error("基本信息获取失败：",e.fillInStackTrace());
    }
  }
}

3.4、编写测试单元

package com.example.mgt.examplemgtservicek8stest;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.example.mgt.examplemgtservicek8stest.rabbitmq.publisher.BasicPublisher;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

/**
 * RabbitMQ 的 JAVA 单元测试类
 * @since 2020/5/15 15:13
 */
@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest
public class RabbitmqTest {
  private static final Logger log = LoggerFactory.getLogger(RabbitmqTest.class);
  @Autowired
  private ObjectMapper objectMapper;
  @Autowired
  private BasicPublisher basicPublisher;

  @Test
  public void test1() throws Exception{
    String msg = "~~~~~~~~~~测试";
    basicPublisher.sendMsg(msg);
  }
}