RabbitMQ 作为常用消息中间件，本文演示如何在 Spring Boot 中快速集成。涵盖环境搭建、依赖配置、消息发送与接收、队列声明、JSON 序列化、手动确认机制、死信队列处理及并发优化等内容。通过 Hello World 示例，展示从 Producer 到 Consumer 的完整流程，并提供 Testcontainers 单元测试方案及常见问题排查指南，帮助开发者掌握 RabbitMQ 核心用法。
在现代分布式系统架构中，消息队列扮演着至关重要的角色。它不仅能够解耦系统组件，还能实现异步处理、流量削峰、可靠投递等多种功能。而在众多消息队列中间件中，RabbitMQ 凭借其稳定性、易用性以及丰富的功能特性，成为 Java 开发者最常选择的消息中间件之一。
今天，我们将从零开始，用 Spring Boot 构建一个最简单的 RabbitMQ "Hello World" 程序。这个程序虽然简单，但却是理解 RabbitMQ 与 Spring Boot 集成机制的基石。通过本文，你将掌握：
RabbitMQ 是一个开源的消息代理（Message Broker）和队列服务器，基于 AMQP（Advanced Message Queuing Protocol）协议实现。它允许应用程序通过'生产者 - 消费者'模型进行异步通信。
在深入代码之前，先了解几个关键术语：
💡 在最简单的'Hello World'场景中，我们通常使用默认的 Direct Exchange 和一个命名队列，无需显式声明 Exchange 或 Binding。
RabbitMQ 官方文档对这些概念有非常清晰的解释，推荐阅读 RabbitMQ Tutorials。
要运行我们的示例，你需要以下环境：
最简单的方式是使用 Docker：
docker run -d --hostname my-rabbit --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management
5672 是 AMQP 协议端口（客户端连接用）15672 是管理界面端口（浏览器访问）启动后，你可以通过 http://localhost:15672 访问管理后台，默认账号密码为 guest/guest。
✅ 如果你不想用 Docker，也可以直接下载安装包：RabbitMQ Download
使用 Spring Initializr 快速生成项目。
选择以下依赖：
或者直接在 pom.xml 中添加：
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
</dependencies>
📌 注意：
spring-boot-starter-amqp是 Spring Boot 对 RabbitMQ 的官方支持模块，内部封装了spring-rabbit，提供了自动配置、模板类、监听器等便利功能。
在 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 是 Spring 提供的用于发送消息的模板类，类似于 RestTemplate 或 JdbcTemplate。
// HelloController.java
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 相同）🔄 注意：在 RabbitMQ 中，默认 Exchange 是一个特殊的 Direct Exchange，它会将消息路由到与 routing key 同名的队列。因此，当我们发送到
"hello.queue"时，RabbitMQ 会尝试将消息投递到名为hello.queue的队列。
消息的消费通过 @RabbitListener 注解实现，非常简洁。
// HelloListener.java
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。
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）该队列。
虽然自动声明很方便，但在生产环境中，建议显式声明队列，以便控制其属性（如持久化、最大长度等）。
// RabbitConfig.java
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 Listener
Producer->>MQ: 发送消息到 hello.queue
MQ->>Consumer: 推送消息
Consumer->>Consumer: 处理消息 (打印日志)
这个流程展示了典型的'点对点'通信模型：一个生产者发送，一个消费者接收。
默认情况下，RabbitTemplate 使用 SimpleMessageConverter，它会将：
String → UTF-8 字节byte[] → 原样传输Serializable 对象 → Java 序列化但 Java 原生序列化存在兼容性问题，且不可读。因此，强烈建议使用 JSON。
// RabbitConfig.java (追加)
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 不仅跨语言友好，还便于调试和监控。
默认情况下，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 崩溃。
即使有手动确认，我们仍可能遇到临时故障（如数据库连接失败）。此时，我们希望消息能重试几次，若仍失败，则转入死信队列（Dead Letter Queue） 供人工处理。
spring:
rabbitmq:
listener:
simple:
retry:
enabled: true
max-attempts: 3
initial-interval: 1000
Spring 会自动在异常时重试最多 3 次，间隔 1 秒。
@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 是外部依赖，但我们仍可以编写集成测试。
@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 容器，确保测试环境与生产一致。
@Component）host、port、username、password 是否正确MessageConverterdurable=true）MessageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT)）📚 更多排错技巧可参考 RabbitMQ Troubleshooting Guide
| 特性 | RabbitMQ | Kafka | RocketMQ |
|---|---|---|---|
| 协议 | AMQP | 自定义 | 自定义 |
| 延迟消息 | 插件支持 | 不支持 | 原生支持 |
| 事务 | 支持 | 不支持 | 支持 |
| 吞吐量 | 中等 | 极高 | 高 |
| 易用性 | ⭐⭐⭐⭐⭐ | ⭐⭐⭐ | ⭐⭐⭐⭐ |
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' 程序，我们掌握了：
虽然只是一个入门示例，但它涵盖了 RabbitMQ 在 Spring Boot 中的核心用法。后续你可以在此基础上扩展：
🌟 记住：消息队列不是银弹，但它是构建弹性、可扩展系统的关键工具。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
查找任何按下的键的javascript键代码、代码、位置和修饰符。 在线工具，Keycode 信息在线工具，online
JavaScript 字符串转义/反转义；Java 风格 \uXXXX（Native2Ascii）编码与解码。 在线工具，Escape 与 Native 编解码在线工具，online
使用 Prettier 在浏览器内格式化 JavaScript 或 HTML 片段。 在线工具，JavaScript / HTML 格式化在线工具，online
Terser 压缩、变量名混淆，或 javascript-obfuscator 高强度混淆（体积会增大）。 在线工具，JavaScript 压缩与混淆在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online