Spring Boot 集成 RabbitMQ 实战指南：消息收发与配置详解

📌 注意：spring-boot-starter-amqp 是 Spring Boot 对 RabbitMQ 的官方支持模块，内部封装了 spring-rabbit，提供了自动配置、模板类、监听器等便利功能。

配置 RabbitMQ 连接

在 application.yml 中添加 RabbitMQ 连接信息：

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /

🔐 Virtual Host（虚拟主机） 是 RabbitMQ 的命名空间机制，类似数据库中的 schema。默认 / 即可。

Spring Boot 会自动读取这些配置，并创建 ConnectionFactory、RabbitTemplate、RabbitAdmin 等核心 Bean。

发送消息：RabbitTemplate

RabbitTemplate 是 Spring 提供的用于发送消息的模板类，类似于 RestTemplate 或 JdbcTemplate。

创建一个 Controller

package com.example.demo;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class HelloController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("/send")
    public String sendMessage(@RequestParam String msg) {
        rabbitTemplate.convertAndSend("hello.queue", msg);
        return "Message sent: " + msg;
    }
}

这里我们调用 convertAndSend(String routingKey, Object message) 方法：

• 第一个参数 "hello.queue" 实际上是 队列名称（在默认 Direct Exchange 下，routing key 与 queue name 相同）
• 第二个参数是消息内容，可以是任意对象，Spring 会自动序列化为字节

🔄 注意：在 RabbitMQ 中，默认 Exchange 是一个特殊的 Direct Exchange，它会将消息路由到与 routing key 同名的队列。因此，当我们发送到 "hello.queue" 时，RabbitMQ 会尝试将消息投递到名为 hello.queue 的队列。

接收消息：@RabbitListener

消息的消费通过 @RabbitListener 注解实现，非常简洁。

创建一个 Listener

package com.example.demo;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class HelloListener {

    @RabbitListener(queues = "hello.queue")
    public void receiveMessage(String message) {
        System.out.println("【收到消息】: " + message);
    }
}

• @RabbitListener(queues = "hello.queue") 表示监听名为 hello.queue 的队列
• 方法参数 String message 会被 Spring 自动反序列化

🧠 Spring 默认使用 SimpleMessageConverter，它支持 String、byte[]、Serializable 对象。对于 JSON，我们可以稍后自定义 MessageConverter。

启动并测试

1. 启动 Spring Boot 应用
2. 访问 http://localhost:8080/send?msg=Hello%20RabbitMQ

你会在控制台看到：

【收到消息】: Hello RabbitMQ

同时，在 RabbitMQ 管理界面（http://localhost:15672）的 Queues 标签页中，可以看到 hello.queue 队列已被自动创建（因为 RabbitAdmin 默认启用自动声明）。

自动声明队列：为什么能成功？

你可能会疑惑：我们并没有显式创建 hello.queue 队列，为什么消息能被成功接收？

这得益于 Spring Boot 的 自动声明机制。

当 @RabbitListener 注解指定了一个队列名，而该队列在 RabbitMQ 中不存在时，Spring 会自动向 Broker 声明（Declare）该队列。

显式声明队列（推荐做法）

虽然自动声明很方便，但在生产环境中，建议显式声明队列，以便控制其属性（如持久化、最大长度等）。

package com.example.demo;

import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {

    @Bean
    public Queue helloQueue() {
        return new Queue("hello.queue", true); // true 表示队列持久化
    }
}

• new Queue("name", durable)：第二个参数表示是否持久化（重启 RabbitMQ 后队列是否保留）
• 其他构造参数还包括：exclusive（排他性）、autoDelete（自动删除）等

💾 持久化很重要！如果队列不持久化，RabbitMQ 重启后队列会消失，导致消息丢失。

消息流转流程图

让我们用 Mermaid 图来可视化整个消息传递过程：

sequenceDiagram
    participant Producer as 生产者 (Controller)
    participant MQ as RabbitMQ
    participant Consumer as 消费者 (@RabbitListener)

    Producer->>MQ: 发送消息到 hello.queue
    MQ->>MQ: 消息入队 (Queue: hello.queue)
    MQ->>Consumer: 推送消息
    Consumer->>Consumer: 处理消息 (打印日志)

这个流程展示了典型的'点对点'通信模型：一个生产者发送，一个消费者接收。

深入理解：消息是如何被序列化的？

默认情况下，RabbitTemplate 使用 SimpleMessageConverter，它会将：

• String → UTF-8 字节
• byte[] → 原样传输
• Serializable 对象 → Java 序列化

但 Java 原生序列化存在兼容性问题，且不可读。因此，强烈建议使用 JSON。

配置 JSON 消息转换器

package com.example.demo;

import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {

    @Bean
    public Queue helloQueue() {
        return new Queue("hello.queue", true);
    }

    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }
}

现在，你可以发送任意 Java 对象：

// 定义消息体
public class Greeting {
    private String content;

    public Greeting(String content) {
        this.content = content;
    }

    public String getContent() { return content; }
    public void setContent(String content) { this.content = content; }
}

// Controller 中
@GetMapping("/sendObj")
public String sendObject() {
    Greeting greeting = new Greeting("Hello from Object!");
    rabbitTemplate.convertAndSend("hello.queue", greeting);
    return "Object sent";
}

对应的 Listener 也要调整：

@RabbitListener(queues = "hello.queue")
public void receiveObject(Greeting greeting) {
    System.out.println("【收到对象】: " + greeting.getContent());
}

✅ 使用 JSON 不仅跨语言友好，还便于调试和监控。

消息确认机制（Acknowledgement）

默认情况下，Spring 的 @RabbitListener 使用 自动确认（Auto Ack） 模式：一旦消息被投递给消费者方法，RabbitMQ 就认为消息已成功处理，并从队列中删除。

但如果消费者在处理过程中崩溃，消息就会丢失！

启用手动确认

在 application.yml 中配置：

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: manual

然后在 Listener 中手动确认：

import com.rabbitmq.client.Channel;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;

@RabbitListener(queues = "hello.queue")
public void receiveMessage(String message, Channel channel,
                           @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
    try {
        System.out.println("处理消息：" + message);
        // 模拟业务逻辑
        if ("error".equals(message)) {
            throw new RuntimeException("模拟异常");
        }
        // 手动确认
        channel.basicAck(deliveryTag, false);
    } catch (Exception e) {
        try {
            // 拒绝消息，不重新入队
            channel.basicNack(deliveryTag, false, false);
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }
}

• basicAck：确认消息处理成功
• basicNack：拒绝消息，第三个参数 requeue 决定是否重新入队

⚠️ 在手动确认模式下，必须显式调用 ack/nack，否则消息会一直'未确认'，占用内存，最终可能导致 RabbitMQ 崩溃。

重试与死信队列（DLQ）

即使有手动确认，我们仍可能遇到临时故障（如数据库连接失败）。此时，我们希望消息能重试几次，若仍失败，则转入死信队列（Dead Letter Queue） 供人工处理。

配置重试机制

spring:
  rabbitmq:
    listener:
      simple:
        retry:
          enabled: true
          max-attempts: 3
          initial-interval: 1000

Spring 会自动在异常时重试最多 3 次，间隔 1 秒。

死信队列配置

1. 声明死信交换机和队列
2. 主队列绑定死信交换机

@Bean
public Queue dlq() {
    return QueueBuilder.durable("hello.dlq").build();
}

@Bean
public DirectExchange dlx() {
    return new DirectExchange("hello.dlx");
}

@Bean
public Binding dlqBinding() {
    return BindingBuilder.bind(dlq()).to(dlx()).with("hello.dlq");
}

@Bean
public Queue helloQueue() {
    return QueueBuilder.durable("hello.queue")
            .withArgument("x-dead-letter-exchange", "hello.dlx")
            .withArgument("x-dead-letter-routing-key", "hello.dlq")
            .build();
}

当消息重试 3 次仍失败，就会被路由到 hello.dlq 队列。

你可以再写一个 Listener 监听 DLQ：

@RabbitListener(queues = "hello.dlq")
public void handleDlq(String message) {
    System.err.println("【死信队列】: " + message);
    // 报警、记录日志、人工干预等
}

性能优化：并发消费者

默认情况下，每个 @RabbitListener 只启动一个消费者线程。如果消息量大，处理慢，会导致积压。

可以通过配置增加并发数：

spring:
  rabbitmq:
    listener:
      simple:
        concurrency: 5
        max-concurrency: 10

• concurrency：初始消费者数量
• max-concurrency：最大消费者数量（动态扩容）

🚀 高并发场景下，合理设置并发数能显著提升吞吐量。

单元测试：如何测试 RabbitMQ？

虽然 RabbitMQ 是外部依赖，但我们仍可以编写集成测试。

使用 Testcontainers（推荐）

@SpringBootTest
@Testcontainers
class RabbitMQIntegrationTest {

    @Container
    static RabbitMQContainer rabbitMQ = new RabbitMQContainer("rabbitmq:3-management");

    @DynamicPropertySource
    static void configureProperties(DynamicPropertyRegistry registry) {
        registry.add("spring.rabbitmq.host", rabbitMQ::getHost);
        registry.add("spring.rabbitmq.port", rabbitMQ::getAmqpPort);
        registry.add("spring.rabbitmq.username", rabbitMQ::getAdminUsername);
        registry.add("spring.rabbitmq.password", rabbitMQ::getAdminPassword);
    }

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    void testSendMessage() throws InterruptedException {
        // 发送消息
        rabbitTemplate.convertAndSend("test.queue", "Hello Test");
        // 等待消费者处理（实际项目中应使用 CountDownLatch 或 Awaitility）
        Thread.sleep(1000);
        // 验证逻辑（例如检查数据库、日志等）
    }
}

🧪 Testcontainers 能在测试时启动真实的 RabbitMQ 容器，确保测试环境与生产一致。

常见问题排查

1. 消息发送成功，但消费者没收到？

• 检查队列名称是否一致
• 查看 RabbitMQ 管理界面，确认消息是否在队列中
• 检查 Listener 是否被 Spring 扫描到（@Component）

2. 启动时报连接拒绝？

• 确认 RabbitMQ 服务已启动
• 检查 host、port、username、password 是否正确
• 防火墙是否放行 5672 端口

3. 消息乱码？

• 确保生产者和消费者使用相同的 MessageConverter
• JSON 场景下，检查对象字段是否匹配

4. 消息丢失？

• 队列是否持久化？（durable=true）
• 消息是否标记为持久化？（MessageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT)）
• 消费者是否启用了手动确认？

RabbitMQ 与其他消息队列对比

RabbitMQ 适合复杂路由、低延迟、高可靠性的场景，如订单处理、通知系统等。

扩展：发布/订阅模式

除了点对点，RabbitMQ 还支持发布/订阅（Pub/Sub）。

示例：广播消息

// 声明 Fanout Exchange
@Bean
public FanoutExchange fanoutExchange() {
    return new FanoutExchange("broadcast.exchange");
}

@Bean
public Queue emailQueue() {
    return new Queue("email.queue");
}

@Bean
public Queue smsQueue() {
    return new Queue("sms.queue");
}

// 绑定
@Bean
public Binding bindEmail() {
    return BindingBuilder.bind(emailQueue()).to(fanoutExchange());
}

@Bean
public Binding bindSms() {
    return BindingBuilder.bind(smsQueue()).to(fanoutExchange());
}

发送消息：

rabbitTemplate.convertAndSend("broadcast.exchange", "", "Hello All!");

两个 Listener 都会收到消息。

总结

通过这个极简的 'Hello World' 程序，我们掌握了：

• Spring Boot 与 RabbitMQ 的无缝集成
• 消息的发送与接收
• 队列的声明与配置
• JSON 序列化、手动确认、重试、死信队列等高级特性

虽然只是一个入门示例，但它涵盖了 RabbitMQ 在 Spring Boot 中的核心用法。后续你可以在此基础上扩展：

• 消息追踪（如集成 Sleuth）
• 监控告警（如 Prometheus + Grafana）
• 集群部署与高可用

🌟 记住：消息队列不是银弹，但它是构建弹性、可扩展系统的关键工具。

Happy Coding！