RabbitMQ 从入门到精通：Spring Boot 实战三部曲（一）—— 基础核心与快速上手

RabbitMQ 从入门到精通：Spring Boot 实战三部曲（一）—— 基础核心与快速上手

专题导读：本系列共三篇，从基础到高级，带你系统掌握 RabbitMQ 在 Spring Boot 项目中的实战应用。

• 第一篇：基础核心与快速上手（本文）
• 第二篇：进阶特性与可靠性保障
• 第三篇：高级应用与性能优化

📖 前言

在当今的分布式系统中，消息队列已成为不可或缺的基础设施。RabbitMQ 作为最流行的消息中间件之一，以其可靠性、灵活性和易用性著称。

本文将从 RabbitMQ 的基础概念出发，结合 Spring Boot 项目，带你快速上手 RabbitMQ 开发，掌握核心概念和实际应用。

学完本文你将掌握：

• ✅ RabbitMQ 的核心概念与应用场景
• ✅ 五种工作模式详解
• ✅ Spring Boot 集成 RabbitMQ 的完整配置
• ✅ 实际业务场景中的消息发送与接收
• ✅ 常见问题的解决方案

一、RabbitMQ 是什么？为什么需要它？

1.1 RabbitMQ 简介

RabbitMQ 是一个开源的消息代理和队列服务器，基于 AMQP（Advanced Message Queuing Protocol）协议实现。

核心特点：

• 🚀 可靠性：支持消息持久化、事务、确认机制
• 🔄 灵活性：支持多种消息路由模式
• 🛡️ 高可用：支持集群、镜像队列
• 🌐 多语言支持：提供多种语言的客户端
• 📊 管理界面：内置 Web 管理控制台

1.2 典型应用场景

┌─────────────────────────────────────────────┐
│ RabbitMQ 应用场景 │
├──────────────┬──────────────────────────────┤
│ 异步处理 │ 注册发送邮件、订单处理 │
│ 应用解耦 │ 微服务间通信、系统拆分 │
│ 流量削峰 │ 秒杀活动、突发流量 │
│ 日志收集 │ 分布式日志聚合 │
│ 任务队列 │ 定时任务、批量处理 │
│ 事件驱动 │ 状态变更通知、数据同步 │
└──────────────┴──────────────────────────────┘

1.3 与其他消息队列对比

二、环境搭建与快速开始

2.1 RabbitMQ 安装

Docker 安装（推荐）

# 拉取镜像
docker pull rabbitmq:3.12-management

# 启动容器
docker run -d –name rabbitmq \\
-p 5672:5672 \\
-p 15672:15672 \\
-e RABBITMQ_DEFAULT_USER=admin \\
-e RABBITMQ_DEFAULT_PASS=admin123 \\
rabbitmq:3.12-management

# 查看日志
docker logs -f rabbitmq

访问管理界面：

• URL: http://localhost:15672
• 用户名: admin
• 密码: admin123

Linux 安装

# Ubuntu/Debian
sudo apt-get install rabbitmq-server

# CentOS/RHEL
sudo yum install rabbitmq-server

# 启动服务
sudo systemctl start rabbitmq-server
sudo systemctl enable rabbitmq-server

# 启用管理插件
sudo rabbitmq-plugins enable rabbitmq_management

2.2 核心概念

┌──────────────────────────────────────────────┐
│ RabbitMQ 核心概念 │
├──────────┬───────────────────────────────────┤
│ Producer │ 消息生产者，发送消息 │
│ Consumer │ 消息消费者，接收并处理消息 │
│ Queue │ 消息队列，存储消息 │
│ Exchange │ 交换机，接收并路由消息 │
│ Binding │ 绑定关系，连接 Exchange 和 Queue │
│ Routing │ 路由键，决定消息路由规则 │
│ Key │ │
└──────────┴───────────────────────────────────┘

消息流转过程：

Producer → Exchange → (Binding + Routing Key) → Queue → Consumer

三、五种工作模式详解

3.1 简单队列模式（Simple Queue）

最简单的模式：一个生产者，一个消费者，一个队列。

Producer → [Queue] → Consumer

Spring Boot 实现

1. Maven 依赖

<dependencies>
<!– Spring Boot Starter AMQP –>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
</dependencies>

2. application.yml 配置

spring:
rabbitmq:
host: localhost
port: 5672
username: admin
password: admin123
virtual-host: /
# 连接池配置
connection-timeout: 15000
# 发布者确认
publisher-confirm-type: correlated
publisher-returns: true
# 消费者配置
listener:
simple:
acknowledge-mode: manual # 手动确认
concurrency: 5 # 最小消费者数量
max-concurrency: 10 # 最大消费者数量
prefetch: 1 # 每次预取消息数

3. 配置类

@Configuration
public class RabbitMQConfig {

/**
* 声明简单队列
*/
@Bean
public Queue simpleQueue() {
// durable: 是否持久化
// exclusive: 是否排他
// autoDelete: 是否自动删除
return new Queue("simple.queue", true, false, false);
}
}

4. 生产者

@Component
@Slf4j
public class SimpleProducer {

@Autowired
private RabbitTemplate rabbitTemplate;

/**
* 发送消息
*/
public void sendMessage(String message) {
rabbitTemplate.convertAndSend("simple.queue", message);
log.info("发送消息: {}", message);
}

/**
* 发送对象消息
*/
public void sendObject(Object obj) {
rabbitTemplate.convertAndSend("simple.queue", obj);
log.info("发送对象消息: {}", obj);
}
}

5. 消费者

@Component
@Slf4j
public class SimpleConsumer {

/**
* 监听简单队列
*/
@RabbitListener(queues = "simple.queue")
public void receiveMessage(String message) {
log.info("收到消息: {}", message);
// 处理业务逻辑
processMessage(message);
}

/**
* 监听对象消息
*/
@RabbitListener(queues = "simple.queue")
public void receiveObject(Message message, Channel channel) throws Exception {
try {
// 反序列化消息
String body = new String(message.getBody(), "UTF-8");
log.info("收到对象消息: {}", body);

// 处理业务逻辑
processMessage(body);

// 手动确认
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

} catch (Exception e) {
log.error("消息处理失败", e);
// 拒绝消息，重新入队
channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
}
}

private void processMessage(String message) {
// 业务处理逻辑
System.out.println("Processing: " + message);
}
}

6. 测试代码

@SpringBootTest
@RunWith(SpringRunner.class)
public class SimpleQueueTest {

@Autowired
private SimpleProducer simpleProducer;

@Test
public void testSendMessage() {
// 发送字符串消息
simpleProducer.sendMessage("Hello RabbitMQ!");

// 发送对象消息
User user = new User();
user.setId(1L);
user.setName("张三");
user.setEmail("zhangsan@example.com");
simpleProducer.sendObject(user);
}
}

3.2 工作队列模式（Work Queue）

多个消费者竞争消费同一个队列的消息，实现负载均衡。

Producer → [Queue] → Consumer1
→ Consumer2
→ Consumer3

实战案例：订单处理

1. 配置类

@Configuration
public class WorkQueueConfig {

/**
* 声明工作队列
*/
@Bean
public Queue workQueue() {
return new Queue("work.queue", true);
}
}

2. 生产者

@Component
@Slf4j
public class OrderProducer {

@Autowired
private RabbitTemplate rabbitTemplate;

/**
* 发送订单消息
*/
public void sendOrder(Order order) {
rabbitTemplate.convertAndSend("work.queue", order);
log.info("发送订单消息: orderId={}", order.getOrderId());
}

/**
* 批量发送订单
*/
public void batchSendOrders(List<Order> orders) {
for (Order order : orders) {
rabbitTemplate.convertAndSend("work.queue", order);
}
log.info("批量发送订单，数量: {}", orders.size());
}
}

3. 多个消费者

@Component
@Slf4j
public class OrderConsumer1 {

@RabbitListener(queues = "work.queue")
public void processOrder(Order order, Channel channel, Message message) throws Exception {
try {
log.info("消费者1处理订单: orderId={}", order.getOrderId());

// 模拟处理耗时
Thread.sleep(1000);

// 处理订单逻辑
processOrderLogic(order);

// 手动确认
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

} catch (Exception e) {
log.error("订单处理失败", e);
channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
}
}

private void processOrderLogic(Order order) {
// 订单处理逻辑
System.out.println("Processing order: " + order.getOrderId());
}
}

@Component
@Slf4j
public class OrderConsumer2 {

@RabbitListener(queues = "work.queue")
public void processOrder(Order order, Channel channel, Message message) throws Exception {
try {
log.info("消费者2处理订单: orderId={}", order.getOrderId());

Thread.sleep(1000);

processOrderLogic(order);

channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

} catch (Exception e) {
log.error("订单处理失败", e);
channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
}
}

private void processOrderLogic(Order order) {
System.out.println("Processing order: " + order.getOrderId());
}
}

4. 测试代码

@SpringBootTest
public class WorkQueueTest {

@Autowired
private OrderProducer orderProducer;

@Test
public void testWorkQueue() throws InterruptedException {
// 发送100个订单
for (int i = 1; i <= 100; i++) {
Order order = new Order();
order.setOrderId((long) i);
order.setAmount(new BigDecimal(i * 100));
order.setCreateTime(new Date());

orderProducer.sendOrder(order);
}

// 等待处理完成
Thread.sleep(30000);
}
}

结果：

• 两个消费者平均分配100个订单
• 每个消费者处理约50个订单
• 实现负载均衡

3.3 发布订阅模式（Publish/Subscribe）

一个生产者发送消息，多个消费者都能收到相同的消息。

Producer → [Fanout Exchange] → Queue1 → Consumer1
→ Queue2 → Consumer2
→ Queue3 → Consumer3

实战案例：用户注册通知

1. 配置类

@Configuration
public class FanoutConfig {

/**
* 声明 Fanout 交换机
*/
@Bean
public FanoutExchange fanoutExchange() {
return new FanoutExchange("user.fanout.exchange");
}

/**
* 声明邮件队列
*/
@Bean
public Queue emailQueue() {
return new Queue("email.queue", true);
}

/**
* 声明短信队列
*/
@Bean
public Queue smsQueue() {
return new Queue("sms.queue", true);
}

/**
* 声明站内信队列
*/
@Bean
public Queue notificationQueue() {
return new Queue("notification.queue", true);
}

/**
* 绑定邮件队列到交换机
*/
@Bean
public Binding emailBinding(Queue emailQueue, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(emailQueue).to(fanoutExchange);
}

/**
* 绑定短信队列到交换机
*/
@Bean
public Binding smsBinding(Queue smsQueue, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(smsQueue).to(fanoutExchange);
}

/**
* 绑定站内信队列到交换机
*/
@Bean
public Binding notificationBinding(Queue notificationQueue, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(notificationQueue).to(fanoutExchange);
}
}

2. 生产者

@Component
@Slf4j
public class UserRegisterProducer {

@Autowired
private RabbitTemplate rabbitTemplate;

/**
* 发送用户注册消息
*/
public void sendUserRegister(User user) {
rabbitTemplate.convertAndSend("user.fanout.exchange", "", user);
log.info("发送用户注册消息: userId={}, username={}",
user.getId(), user.getUsername());
}
}

3. 消费者

@Component
@Slf4j
public class EmailConsumer {

@RabbitListener(queues = "email.queue")
public void sendEmail(User user) {
log.info("发送邮件通知: userId={}, email={}",
user.getId(), user.getEmail());

// 调用邮件服务
sendWelcomeEmail(user.getEmail(), user.getUsername());
}

private void sendWelcomeEmail(String email, String username) {
// 发送邮件逻辑
System.out.println("Sending welcome email to: " + email);
}
}

@Component
@Slf4j
public class SmsConsumer {

@RabbitListener(queues = "sms.queue")
public void sendSms(User user) {
log.info("发送短信通知: userId={}, phone={}",
user.getId(), user.getPhone());

// 调用短信服务
sendWelcomeSms(user.getPhone(), user.getUsername());
}

private void sendWelcomeSms(String phone, String username) {
// 发送短信逻辑
System.out.println("Sending welcome SMS to: " + phone);
}
}

@Component
@Slf4j
public class NotificationConsumer {

@RabbitListener(queues = "notification.queue")
public void sendNotification(User user) {
log.info("发送站内信通知: userId={}", user.getId());

// 创建站内信
createNotification(user.getId(), "欢迎注册");
}

private void createNotification(Long userId, String content) {
// 创建站内信逻辑
System.out.println("Creating notification for user: " + userId);
}
}

4. 测试代码

@SpringBootTest
public class FanoutTest {

@Autowired
private UserRegisterProducer producer;

@Test
public void testFanout() {
User user = new User();
user.setId(1L);
user.setUsername("张三");
user.setEmail("zhangsan@example.com");
user.setPhone("13800138000");

producer.sendUserRegister(user);

// 三个消费者都会收到消息
}
}

3.4 路由模式（Routing）

根据 routing key 将消息路由到不同的队列。

Producer → [Direct Exchange] → Routing Key="error" → Error Queue
→ Routing Key="warning" → Warning Queue
→ Routing Key="info" → Info Queue

实战案例：日志分级处理

1. 配置类

@Configuration
public class DirectConfig {

/**
* 声明 Direct 交换机
*/
@Bean
public DirectExchange directExchange() {
return new DirectExchange("log.direct.exchange");
}

/**
* 声明错误日志队列
*/
@Bean
public Queue errorQueue() {
return new Queue("log.error.queue", true);
}

/**
* 声明警告日志队列
*/
@Bean
public Queue warningQueue() {
return new Queue("log.warning.queue", true);
}

/**
* 声明信息日志队列
*/
@Bean
public Queue infoQueue() {
return new Queue("log.info.queue", true);
}

/**
* 绑定错误日志队列
*/
@Bean
public Binding errorBinding(Queue errorQueue, DirectExchange directExchange) {
return BindingBuilder.bind(errorQueue)
.to(directExchange)
.with("error");
}

/**
* 绑定警告日志队列
*/
@Bean
public Binding warningBinding(Queue warningQueue, DirectExchange directExchange) {
return BindingBuilder.bind(warningQueue)
.to(directExchange)
.with("warning");
}

/**
* 绑定信息日志队列
*/
@Bean
public Binding infoBinding(Queue infoQueue, DirectExchange directExchange) {
return BindingBuilder.bind(infoQueue)
.to(directExchange)
.with("info");
}
}

2. 消息实体

@Data
@AllArgsConstructor
@NoArgsConstructor
public class LogMessage implements Serializable {

private static final long serialVersionUID = 1L;

private String level; // 日志级别
private String message; // 日志内容
private String serviceName;// 服务名称
private Long timestamp; // 时间戳
}

3. 生产者

@Component
@Slf4j
public class LogProducer {

@Autowired
private RabbitTemplate rabbitTemplate;

/**
* 发送错误日志
*/
public void sendErrorLog(String message, String serviceName) {
LogMessage logMessage = new LogMessage("error", message, serviceName, System.currentTimeMillis());
rabbitTemplate.convertAndSend("log.direct.exchange", "error", logMessage);
log.info("发送错误日志: {}", message);
}

/**
* 发送警告日志
*/
public void sendWarningLog(String message, String serviceName) {
LogMessage logMessage = new LogMessage("warning", message, serviceName, System.currentTimeMillis());
rabbitTemplate.convertAndSend("log.direct.exchange", "warning", logMessage);
log.info("发送警告日志: {}", message);
}

/**
* 发送信息日志
*/
public void sendInfoLog(String message, String serviceName) {
LogMessage logMessage = new LogMessage("info", message, serviceName, System.currentTimeMillis());
rabbitTemplate.convertAndSend("log.direct.exchange", "info", logMessage);
log.info("发送信息日志: {}", message);
}
}

4. 消费者

@Component
@Slf4j
public class ErrorLogConsumer {

@RabbitListener(queues = "log.error.queue")
public void consumeErrorLog(LogMessage logMessage) {
log.error("【错误日志】服务: {}, 内容: {}",
logMessage.getServiceName(), logMessage.getMessage());

// 保存错误日志到数据库
saveErrorLog(logMessage);

// 发送告警通知
sendAlert(logMessage);
}

private void saveErrorLog(LogMessage logMessage) {
// 保存逻辑
}

private void sendAlert(LogMessage logMessage) {
// 告警逻辑
}
}

@Component
@Slf4j
public class WarningLogConsumer {

@RabbitListener(queues = "log.warning.queue")
public void consumeWarningLog(LogMessage logMessage) {
log.warn("【警告日志】服务: {}, 内容: {}",
logMessage.getServiceName(), logMessage.getMessage());

// 保存警告日志
saveWarningLog(logMessage);
}

private void saveWarningLog(LogMessage logMessage) {
// 保存逻辑
}
}

@Component
@Slf4j
public class InfoLogConsumer {

@RabbitListener(queues = "log.info.queue")
public void consumeInfoLog(LogMessage logMessage) {
log.info("【信息日志】服务: {}, 内容: {}",
logMessage.getServiceName(), logMessage.getMessage());

// 保存信息日志
saveInfoLog(logMessage);
}

private void saveInfoLog(LogMessage logMessage) {
// 保存逻辑
}
}

5. 测试代码

@SpringBootTest
public class DirectTest {

@Autowired
private LogProducer logProducer;

@Test
public void testDirect() {
// 发送不同级别的日志
logProducer.sendErrorLog("数据库连接失败", "order-service");
logProducer.sendWarningLog("内存使用率超过80%", "user-service");
logProducer.sendInfoLog("用户登录成功", "auth-service");

// 每条日志会被路由到对应的队列
}
}

3.5 主题模式（Topic）

根据 routing key 的模式匹配进行路由，更灵活的路由方式。

Producer → [Topic Exchange] → *.error.* → Error Queue
→ order.# → Order Queue
→ user.*.create → UserCreate Queue

匹配规则：

• *：匹配一个单词
• #：匹配零个或多个单词

实战案例：电商消息路由

1. 配置类

@Configuration
public class TopicConfig {

/**
* 声明 Topic 交换机
*/
@Bean
public TopicExchange topicExchange() {
return new TopicExchange("ecommerce.topic.exchange");
}

/**
* 声明订单队列
*/
@Bean
public Queue orderQueue() {
return new Queue("order.queue", true);
}

/**
* 声明用户队列
*/
@Bean
public Queue userQueue() {
return new Queue("user.queue", true);
}

/**
* 声明所有消息队列
*/
@Bean
public Queue allQueue() {
return new Queue("all.queue", true);
}

/**
* 绑定订单队列（匹配 order.*）
*/
@Bean
public Binding orderBinding(Queue orderQueue, TopicExchange topicExchange) {
return BindingBuilder.bind(orderQueue)
.to(topicExchange)
.with("order.*");
}

/**
* 绑定用户队列（匹配 user.#）
*/
@Bean
public Binding userBinding(Queue userQueue, TopicExchange topicExchange) {
return BindingBuilder.bind(userQueue)
.to(topicExchange)
.with("user.#");
}

/**
* 绑定所有消息队列（匹配 #）
*/
@Bean
public Binding allBinding(Queue allQueue, TopicExchange topicExchange) {
return BindingBuilder.bind(allQueue)
.to(topicExchange)
.with("#");
}
}

2. 生产者

@Component
@Slf4j
public class EcommerceProducer {

@Autowired
private RabbitTemplate rabbitTemplate;

/**
* 发送订单创建消息
*/
public void sendOrderCreate(Order order) {
rabbitTemplate.convertAndSend("ecommerce.topic.exchange",
"order.create", order);
log.info("发送订单创建消息: orderId={}", order.getOrderId());
}

/**
* 发送订单支付消息
*/
public void sendOrderPay(Order order) {
rabbitTemplate.convertAndSend("ecommerce.topic.exchange",
"order.pay", order);
log.info("发送订单支付消息: orderId={}", order.getOrderId());
}

/**
* 发送用户注册消息
*/
public void sendUserRegister(User user) {
rabbitTemplate.convertAndSend("ecommerce.topic.exchange",
"user.register", user);
log.info("发送用户注册消息: userId={}", user.getId());
}

/**
* 发送用户修改消息
*/
public void sendUserUpdate(User user) {
rabbitTemplate.convertAndSend("ecommerce.topic.exchange",
"user.profile.update", user);
log.info("发送用户修改消息: userId={}", user.getId());
}
}

3. 消费者

@Component
@Slf4j
public class OrderConsumer {

@RabbitListener(queues = "order.queue")
public void processOrder(Message message, String routingKey) {
log.info("订单消费者收到消息: routingKey={}", routingKey);
// 处理订单相关逻辑
}
}

@Component
@Slf4j
public class UserConsumer {

@RabbitListener(queues = "user.queue")
public void processUser(Message message, String routingKey) {
log.info("用户消费者收到消息: routingKey={}", routingKey);
// 处理用户相关逻辑
}
}

@Component
@Slf4j
public class AllConsumer {

@RabbitListener(queues = "all.queue")
public void processAll(Message message, String routingKey) {
log.info("全局消费者收到消息: routingKey={}", routingKey);
// 记录所有消息
}
}

4. 测试代码

@SpringBootTest
public class TopicTest {

@Autowired
private EcommerceProducer producer;

@Test
public void testTopic() {
// 订单消息
Order order1 = new Order();
order1.setOrderId(1L);
producer.sendOrderCreate(order1);

Order order2 = new Order();
order2.setOrderId(2L);
producer.sendOrderPay(order2);

// 用户消息
User user1 = new User();
user1.setId(1L);
producer.sendUserRegister(user1);

User user2 = new User();
user2.setId(2L);
producer.sendUserUpdate(user2);

/*
* 路由结果：
* order.create → order.queue + all.queue
* order.pay → order.queue + all.queue
* user.register → user.queue + all.queue
* user.profile.update → user.queue + all.queue
*/
}
}

四、Spring Boot 集成最佳实践

4.1 完整配置类

@Configuration
@Slf4j
public class RabbitMQCompleteConfig {

@Value("${spring.rabbitmq.host}")
private String host;

@Value("${spring.rabbitmq.port}")
private int port;

@Value("${spring.rabbitmq.username}")
private String username;

@Value("${spring.rabbitmq.password}")
private String password;

/**
* 连接工厂
*/
@Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory factory = new CachingConnectionFactory();
factory.setHost(host);
factory.setPort(port);
factory.setUsername(username);
factory.setPassword(password);
factory.setVirtualHost("/");

// 开启发布者确认
factory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
factory.setPublisherReturns(true);

return factory;
}

/**
* RabbitTemplate
*/
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate template = new RabbitTemplate(connectionFactory);

// 消息序列化
template.setMessageConverter(new Jackson2JsonMessageConverter());

// 强制返回
template.setMandatory(true);

// 确认回调
template.setConfirmCallback((correlationData, ack, cause) -> {
if (ack) {
log.info("消息发送成功: {}", correlationData);
} else {
log.error("消息发送失败: {}, 原因: {}", correlationData, cause);
}
});

// 返回回调
template.setReturnsCallback(returned -> {
log.error("消息返回: {}, 路由键: {}, 交换机: {}",
new String(returned.getMessage().getBody()),
returned.getRoutingKey(),
returned.getExchange());
});

return template;
}

/**
* 消息转换器
*/
@Bean
public MessageConverter messageConverter() {
return new Jackson2JsonMessageConverter();
}
}

4.2 通用工具类

@Component
@Slf4j
public class RabbitMQUtil {

@Autowired
private RabbitTemplate rabbitTemplate;

/**
* 发送消息到指定队列
*/
public void sendToQueue(String queueName, Object message) {
rabbitTemplate.convertAndSend(queueName, message);
log.info("发送消息到队列: {}, 消息: {}", queueName, message);
}

/**
* 发送消息到交换机
*/
public void sendToExchange(String exchange, String routingKey, Object message) {
rabbitTemplate.convertAndSend(exchange, routingKey, message);
log.info("发送消息到交换机: {}, 路由键: {}", exchange, routingKey);
}

/**
* 延迟发送消息
*/
public void sendWithDelay(String queueName, Object message, long delayMillis) {
MessagePostProcessor processor = msg -> {
msg.getMessageProperties().setDelay((int) delayMillis);
return msg;
};
rabbitTemplate.convertAndSend(queueName, message, processor);
log.info("延迟发送消息: {}ms", delayMillis);
}
}

4.3 消息实体规范

@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
public class BaseMessage implements Serializable {

private static final long serialVersionUID = 1L;

/**
* 消息ID
*/
private String messageId;

/**
* 消息类型
*/
private String messageType;

/**
* 消息内容
*/
private Object data;

/**
* 创建时间
*/
private Long timestamp;

/**
* 版本号
*/
private String version;

@PrePersist
public void prePersist() {
if (this.messageId == null) {
this.messageId = UUID.randomUUID().toString();
}
if (this.timestamp == null) {
this.timestamp = System.currentTimeMillis();
}
if (this.version == null) {
this.version = "1.0";
}
}
}

五、实战项目：电商订单系统

5.1 项目结构

ecommerce-order/
├── config/
│ └── RabbitMQConfig.java
├── producer/
│ ├── OrderProducer.java
│ └── PaymentProducer.java
├── consumer/
│ ├── OrderConsumer.java
│ ├── PaymentConsumer.java
│ └── InventoryConsumer.java
├── service/
│ ├── OrderService.java
│ └── PaymentService.java
└── model/
├── Order.java
└── Payment.java

5.2 核心代码实现

订单创建流程：

@Service
@Slf4j
public class OrderService {

@Autowired
private OrderProducer orderProducer;

@Autowired
private OrderMapper orderMapper;

/**
* 创建订单
*/
@Transactional
public Long createOrder(OrderRequest request) {
// 1. 创建订单
Order order = new Order();
order.setUserId(request.getUserId());
order.setProductId(request.getProductId());
order.setQuantity(request.getQuantity());
order.setAmount(request.getAmount());
order.setStatus(OrderStatus.CREATED);
order.setCreateTime(new Date());

orderMapper.insert(order);

// 2. 发送订单创建消息
orderProducer.sendOrderCreated(order);

log.info("订单创建成功: orderId={}", order.getId());
return order.getId();
}
}

订单消费者：

@Component
@Slf4j
public class OrderConsumer {

@Autowired
private InventoryService inventoryService;

@Autowired
private NotificationService notificationService;

@RabbitListener(queues = "order.created.queue")
public void handleOrderCreated(Order order, Channel channel, Message message) throws Exception {
try {
log.info("处理订单创建: orderId={}", order.getId());

// 1. 扣减库存
boolean success = inventoryService.deductStock(
order.getProductId(),
order.getQuantity()
);

if (!success) {
throw new RuntimeException("库存不足");
}

// 2. 更新订单状态
updateOrderStatus(order.getId(), OrderStatus.PAID);

// 3. 发送通知
notificationService.sendOrderNotification(order);

// 手动确认
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

} catch (Exception e) {
log.error("订单处理失败", e);
// 拒绝消息，不重新入队（进入死信队列）
channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
}
}

private void updateOrderStatus(Long orderId, OrderStatus status) {
// 更新订单状态
}
}

六、常见问题与解决方案

6.1 消息丢失问题

解决方案：开启持久化和确认机制

spring:
rabbitmq:
# 发布者确认
publisher-confirm-type: correlated
publisher-returns: true
# 消费者手动确认
listener:
simple:
acknowledge-mode: manual

// 队列持久化
@Bean
public Queue queue() {
return new Queue("queue.name", true); // durable=true
}

// 交换机持久化
@Bean
public DirectExchange exchange() {
return new DirectExchange("exchange.name", true, false);
}

// 消息持久化
rabbitTemplate.convertAndSend("exchange", "routingKey", message,
msg -> {
msg.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
return msg;
});

6.2 消息重复消费

解决方案：消息幂等性处理

@Component
public class IdempotentConsumer {

@Autowired
private RedisTemplate<String, Object> redisTemplate;

@RabbitListener(queues = "order.queue")
public void consume(Order order, Channel channel, Message message) throws Exception {
String messageId = order.getMessageId();
String key = "message:processed:" + messageId;

try {
// 检查是否已处理
Boolean processed = redisTemplate.opsForValue()
.setIfAbsent(key, "1", 24, TimeUnit.HOURS);

if (Boolean.FALSE.equals(processed)) {
log.warn("消息已处理，跳过: {}", messageId);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
return;
}

// 处理业务逻辑
processOrder(order);

// 确认消息
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

} catch (Exception e) {
log.error("消息处理失败", e);
channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
}
}
}

6.3 消息积压问题

解决方案：增加消费者数量

spring:
rabbitmq:
listener:
simple:
concurrency: 10 # 增加并发消费者数量
max-concurrency: 50
prefetch: 10 # 增加预取数量

七、总结与展望

7.1 本文要点回顾

✅ RabbitMQ 基础概念：理解核心组件和工作原理
✅ 五种工作模式：掌握 Simple、Work、Fanout、Direct、Topic
✅ Spring Boot 集成：完成配置类和工具类封装
✅ 实战案例：电商订单系统完整实现
✅ 常见问题：消息丢失、重复消费、积压的解决方案

7.2 下篇预告

在下一篇文章《RabbitMQ 从入门到精通：Spring Boot 实战三部曲（二）—— 进阶特性与可靠性保障》中，我们将深入探讨：

• 🔒 消息可靠性保障：确认机制、持久化、事务
• 💀 死信队列：处理失败消息的完整方案
• ⏰ 延迟队列：实现定时任务的多种方式
• 🔄 消息幂等性：保证消息不被重复消费
• 📊 监控与管理：RabbitMQ Management 深度使用
• 🛡️ 高可用架构：镜像队列与集群

7.3 学习建议

📚 参考资料

• RabbitMQ 官方文档：https://www.rabbitmq.com/documentation.html
• Spring AMQP 文档：https://spring.io/projects/spring-amqp
• 《RabbitMQ 实战指南》

觉得有用？欢迎点赞、收藏、转发！
下一篇更精彩，敬请期待！ 🚀

系列文章：

• [第一篇] RabbitMQ 从入门到精通：Spring Boot 实战三部曲（一）—— 基础核心与快速上手
• [第二篇] RabbitMQ 从入门到精通：Spring Boot 实战三部曲（二）—— 进阶特性与可靠性保障
• [第三篇] RabbitMQ 从入门到精通：Spring Boot 实战三部曲（三）—— 高级应用与性能优化