SpringBoot 整合 RabbitMQ：Spring AMQP

spring-boot-starter-amqp 是 Spring Boot 提供的一个启动器（Starter），它位于 Spring Boot 的生态体系中，旨在简化与 AMQP（Advanced Message Queuing Protocol，高级消息队列协议）消息中间件的集成开发过程。 spring-boot-starter-amqp 包含了与 AMQP 通信所需的各类库和配置，借助它，开发者能够轻松地在 Spring Boot 项目里使用 AMQP 协议与消息队列进行交互。

环境集成

引入依赖：

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

在 Spring 配置文件中指定用户名、密码、主机地址、端口：

spring:rabbitmq:username: guestpassword: guesthost: 192.168.3.103port: 5672

测试是否集成成功：

• 定义配置类：

  package test;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;@Configuration
  public class RabbitMQConfig {protected static final String EXCHANGE_NAME = "test.direct";protected static final String QUEUE_NAME = "test.queue";protected static final String ROUTING_KEY = "test.routing.key";@Beanpublic Queue queue() {return new Queue(QUEUE_NAME, true);}@Beanpublic DirectExchange exchange() {return new DirectExchange(EXCHANGE_NAME, true, false);}@Beanpublic Binding binding(Queue queue, DirectExchange exchange) {return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);}}
  

• 发送一条消息：

  package test;import org.springframework.amqp.rabbit.core.RabbitTemplate;
  import org.springframework.boot.SpringApplication;
  import org.springframework.boot.autoconfigure.SpringBootApplication;
  import org.springframework.context.ConfigurableApplicationContext;@SpringBootApplication
  public class SpringBootApp {public static void main(String[] args) {ConfigurableApplicationContext context = SpringApplication.run(SpringBootApp.class, args);RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME, RabbitMQConfig.ROUTING_KEY, "hello");}}
  

• 接收消息：

  package test;import org.springframework.amqp.rabbit.annotation.RabbitListener;
  import org.springframework.stereotype.Service;@Service
  public class RabbitMQConsumer {@RabbitListener(queues = RabbitMQConfig.QUEUE_NAME)public void receiveMessage(String message) {System.out.println("Received message: " + message);}}
  

• 输出结果：

  

声明队列

Queue 用于声明队列，可以通过 QueueBuilder 构建：

@Bean
public Queue simpleQueue() {// 创建一个名为simple-queue的非持久化（nonDurable指定队列为非持久化）的队列return QueueBuilder.nonDurable("simple-queue").build();
}

或：

@Bean
public Queue simpleQueue() {// 创建一个名为simple-queue的持久化队列return new Queue("simple-queue");
}

声明交换机

Exchange 用于声明交换机，可以通过工厂类 ExchangeBuilder 构建（默认创建持久化的交换机）：

@Bean
public DirectExchange workQueueDirectExchange(){return ExchangeBuilder.directExchange("work-queue-exchange").build();
}

或：

@Bean
public DirectExchange simpleQueueDirectExchange(){return new DirectExchange("simple-queue-exchange");
}

声明绑定

Binding 用于声明队列和交换机的绑定关系，可以用工厂类 BindingBuilder 构建

@Bean
public Binding workQueueBinding(){return BindingBuilder.bind(workQueue()).to(workQueueDirectExchange()).withQueueName();
}

消息格式转换

在默认情况下，如果生产者与消费者通过 Spring AMQP 发送非 Java 的 String 类型的消息（如 HashMap 类型）。Spring AMQP 会对消息对象进行序列化处理。org.springframework.amqp.support.converter.MessageConverter 接口负责 Spring AMQP 对消息对象的处理，而其默认实现为 SimpleMessageConverter 。SimpleMessageConverter 为基于 JDK 的 ObjectOutputStream 完成序列化。

JDK 序列化存在以下问题：

• 存在安全风险：攻击者可以发送恶意的序列化消息，使服务器在进行反序列化过程中进行违规操作。
• 消息内容过长：原本简短的消息内容由于被序列化的原因导致序列化消息内容比原来的消息内容长很多。
• 消息可读性差：在 RabbitMQ 控制台展示的是序列化后的消息，而非消息原本的格式。

为了避免上述问题，可以通过在 Spring 容器中注入 Jackson2JsonMessageConverter 的实现（在生产者与消费者中都需要注入）：

@Bean
public Jackson2JsonMessageConverter jackson2JsonMessageConverter(){return new Jackson2JsonMessageConverter();
}

注入 Jackson2JsonMessageConverter 后，消息以 JSON 的格式传输，并在 RabbitMQ 控制台也展示 JSON 格式的原消息，大大提高了可读性。