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 典型应用场景

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

1.3 与其他消息队列对比

二、环境搭建与快速开始

2.1 RabbitMQ 安装

Docker 安装（推荐）

1# 拉取镜像
2docker pull rabbitmq:3.12-management
3
4# 启动容器
5docker run -d --name rabbitmq \
6  -p 5672:5672 \
7  -p 15672:15672 \
8  -e RABBITMQ_DEFAULT_USER=admin \
9  -e RABBITMQ_DEFAULT_PASS=admin123 \
10  rabbitmq:3.12-management
11
12# 查看日志
13docker logs -f rabbitmq
14

访问管理界面：

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

Linux 安装

1# Ubuntu/Debian
2sudo apt-get install rabbitmq-server
3
4# CentOS/RHEL
5sudo yum install rabbitmq-server
6
7# 启动服务
8sudo systemctl start rabbitmq-server
9sudo systemctl enable rabbitmq-server
10
11# 启用管理插件
12sudo rabbitmq-plugins enable rabbitmq_management
13

2.2 核心概念

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

消息流转过程：

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

三、五种工作模式详解

3.1 简单队列模式（Simple Queue）

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

1Producer → [Queue] → Consumer
2

Spring Boot 实现

1. Maven 依赖

1<dependencies>
2    <!-- Spring Boot Starter AMQP -->
3    <dependency>
4        <groupId>org.springframework.boot</groupId>
5        <artifactId>spring-boot-starter-amqp</artifactId>
6    </dependency>
7</dependencies>
8

2. application.yml 配置

1spring:
2  rabbitmq:
3    host: localhost
4    port: 5672
5    username: admin
6    password: admin123
7    virtual-host: /
8    # 连接池配置
9    connection-timeout: 15000
10    # 发布者确认
11    publisher-confirm-type: correlated
12    publisher-returns: true
13    # 消费者配置
14    listener:
15      simple:
16        acknowledge-mode: manual  # 手动确认
17        concurrency: 5            # 最小消费者数量
18        max-concurrency: 10       # 最大消费者数量
19        prefetch: 1               # 每次预取消息数
20

3. 配置类

1@Configuration
2public class RabbitMQConfig {
3    
4    /**
5     * 声明简单队列
6     */
7    @Bean
8    public Queue simpleQueue() {
9        // durable: 是否持久化
10        // exclusive: 是否排他
11        // autoDelete: 是否自动删除
12        return new Queue("simple.queue", true, false, false);
13    }
14}
15

4. 生产者

1@Component
2@Slf4j
3public class SimpleProducer {
4    
5    @Autowired
6    private RabbitTemplate rabbitTemplate;
7
8    /**
9     * 发送消息
10     */
11    public void sendMessage(String message) {
12        rabbitTemplate.convertAndSend("simple.queue", message);
13        log.info("发送消息: {}", message);
14    }
15
16    /**
17     * 发送对象消息
18     */
19    public void sendObject(Object obj) {
20        rabbitTemplate.convertAndSend("simple.queue", obj);
21        log.info("发送对象消息: {}", obj);
22    }
23}
24

5. 消费者

1@Component
2@Slf4j
3public class SimpleConsumer {
4    
5    /**
6     * 监听简单队列
7     */
8    @RabbitListener(queues = "simple.queue")
9    public void receiveMessage(String message) {
10        log.info("收到消息: {}", message);
11        // 处理业务逻辑
12        processMessage(message);
13    }
14
15    /**
16     * 监听对象消息
17     */
18    @RabbitListener(queues = "simple.queue")
19    public void receiveObject(Message message, Channel channel) throws Exception {
20        try {
21            // 反序列化消息
22            String body = new String(message.getBody(), "UTF-8");
23            log.info("收到对象消息: {}", body);
24            
25            // 处理业务逻辑
26            processMessage(body);
27            
28            // 手动确认
29            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
30            
31        } catch (Exception e) {
32            log.error("消息处理失败", e);
33            // 拒绝消息，重新入队
34            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
35        }
36    }
37
38    private void processMessage(String message) {
39        // 业务处理逻辑
40        System.out.println("Processing: " + message);
41    }
42}
43

6. 测试代码

1@SpringBootTest
2@RunWith(SpringRunner.class)
3public class SimpleQueueTest {
4    
5    @Autowired
6    private SimpleProducer simpleProducer;
7
8    @Test
9    public void testSendMessage() {
10        // 发送字符串消息
11        simpleProducer.sendMessage("Hello RabbitMQ!");
12        
13        // 发送对象消息
14        User user = new User();
15        user.setId(1L);
16        user.setName("张三");
17        user.setEmail("zhangsan@example.com");
18        simpleProducer.sendObject(user);
19    }
20}
21

3.2 工作队列模式（Work Queue）

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

1Producer → [Queue] → Consumer1
2                   → Consumer2
3                   → Consumer3
4

实战案例：订单处理

1. 配置类

1@Configuration
2public class WorkQueueConfig {
3    
4    /**
5     * 声明工作队列
6     */
7    @Bean
8    public Queue workQueue() {
9        return new Queue("work.queue", true);
10    }
11}
12

2. 生产者

1@Component
2@Slf4j
3public class OrderProducer {
4    
5    @Autowired
6    private RabbitTemplate rabbitTemplate;
7
8    /**
9     * 发送订单消息
10     */
11    public void sendOrder(Order order) {
12        rabbitTemplate.convertAndSend("work.queue", order);
13        log.info("发送订单消息: orderId={}", order.getOrderId());
14    }
15
16    /**
17     * 批量发送订单
18     */
19    public void batchSendOrders(List<Order> orders) {
20        for (Order order : orders) {
21            rabbitTemplate.convertAndSend("work.queue", order);
22        }
23        log.info("批量发送订单，数量: {}", orders.size());
24    }
25}
26

3. 多个消费者

1@Component
2@Slf4j
3public class OrderConsumer1 {
4    
5    @RabbitListener(queues = "work.queue")
6    public void processOrder(Order order, Channel channel, Message message) throws Exception {
7        try {
8            log.info("消费者1处理订单: orderId={}", order.getOrderId());
9            
10            // 模拟处理耗时
11            Thread.sleep(1000);
12            
13            // 处理订单逻辑
14            processOrderLogic(order);
15            
16            // 手动确认
17            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
18            
19        } catch (Exception e) {
20            log.error("订单处理失败", e);
21            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
22        }
23    }
24
25    private void processOrderLogic(Order order) {
26        // 订单处理逻辑
27        System.out.println("Processing order: " + order.getOrderId());
28    }
29}
30
31@Component
32@Slf4j
33public class OrderConsumer2 {
34    
35    @RabbitListener(queues = "work.queue")
36    public void processOrder(Order order, Channel channel, Message message) throws Exception {
37        try {
38            log.info("消费者2处理订单: orderId={}", order.getOrderId());
39            
40            Thread.sleep(1000);
41            
42            processOrderLogic(order);
43            
44            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
45            
46        } catch (Exception e) {
47            log.error("订单处理失败", e);
48            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
49        }
50    }
51
52    private void processOrderLogic(Order order) {
53        System.out.println("Processing order: " + order.getOrderId());
54    }
55}
56

4. 测试代码

1@SpringBootTest
2public class WorkQueueTest {
3    
4    @Autowired
5    private OrderProducer orderProducer;
6
7    @Test
8    public void testWorkQueue() throws InterruptedException {
9        // 发送100个订单
10        for (int i = 1; i <= 100; i++) {
11            Order order = new Order();
12            order.setOrderId((long) i);
13            order.setAmount(new BigDecimal(i * 100));
14            order.setCreateTime(new Date());
15            
16            orderProducer.sendOrder(order);
17        }
18        
19        // 等待处理完成
20        Thread.sleep(30000);
21    }
22}
23

结果：

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

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

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

1Producer → [Fanout Exchange] → Queue1 → Consumer1
2                              → Queue2 → Consumer2
3                              → Queue3 → Consumer3
4

实战案例：用户注册通知

1. 配置类

1@Configuration
2public class FanoutConfig {
3    
4    /**
5     * 声明 Fanout 交换机
6     */
7    @Bean
8    public FanoutExchange fanoutExchange() {
9        return new FanoutExchange("user.fanout.exchange");
10    }
11
12    /**
13     * 声明邮件队列
14     */
15    @Bean
16    public Queue emailQueue() {
17        return new Queue("email.queue", true);
18    }
19
20    /**
21     * 声明短信队列
22     */
23    @Bean
24    public Queue smsQueue() {
25        return new Queue("sms.queue", true);
26    }
27
28    /**
29     * 声明站内信队列
30     */
31    @Bean
32    public Queue notificationQueue() {
33        return new Queue("notification.queue", true);
34    }
35
36    /**
37     * 绑定邮件队列到交换机
38     */
39    @Bean
40    public Binding emailBinding(Queue emailQueue, FanoutExchange fanoutExchange) {
41        return BindingBuilder.bind(emailQueue).to(fanoutExchange);
42    }
43
44    /**
45     * 绑定短信队列到交换机
46     */
47    @Bean
48    public Binding smsBinding(Queue smsQueue, FanoutExchange fanoutExchange) {
49        return BindingBuilder.bind(smsQueue).to(fanoutExchange);
50    }
51
52    /**
53     * 绑定站内信队列到交换机
54     */
55    @Bean
56    public Binding notificationBinding(Queue notificationQueue, FanoutExchange fanoutExchange) {
57        return BindingBuilder.bind(notificationQueue).to(fanoutExchange);
58    }
59}
60

2. 生产者

1@Component
2@Slf4j
3public class UserRegisterProducer {
4    
5    @Autowired
6    private RabbitTemplate rabbitTemplate;
7
8    /**
9     * 发送用户注册消息
10     */
11    public void sendUserRegister(User user) {
12        rabbitTemplate.convertAndSend("user.fanout.exchange", "", user);
13        log.info("发送用户注册消息: userId={}, username={}", 
14            user.getId(), user.getUsername());
15    }
16}
17

3. 消费者

1@Component
2@Slf4j
3public class EmailConsumer {
4    
5    @RabbitListener(queues = "email.queue")
6    public void sendEmail(User user) {
7        log.info("发送邮件通知: userId={}, email={}", 
8            user.getId(), user.getEmail());
9        
10        // 调用邮件服务
11        sendWelcomeEmail(user.getEmail(), user.getUsername());
12    }
13
14    private void sendWelcomeEmail(String email, String username) {
15        // 发送邮件逻辑
16        System.out.println("Sending welcome email to: " + email);
17    }
18}
19
20@Component
21@Slf4j
22public class SmsConsumer {
23    
24    @RabbitListener(queues = "sms.queue")
25    public void sendSms(User user) {
26        log.info("发送短信通知: userId={}, phone={}", 
27            user.getId(), user.getPhone());
28        
29        // 调用短信服务
30        sendWelcomeSms(user.getPhone(), user.getUsername());
31    }
32
33    private void sendWelcomeSms(String phone, String username) {
34        // 发送短信逻辑
35        System.out.println("Sending welcome SMS to: " + phone);
36    }
37}
38
39@Component
40@Slf4j
41public class NotificationConsumer {
42    
43    @RabbitListener(queues = "notification.queue")
44    public void sendNotification(User user) {
45        log.info("发送站内信通知: userId={}", user.getId());
46        
47        // 创建站内信
48        createNotification(user.getId(), "欢迎注册");
49    }
50
51    private void createNotification(Long userId, String content) {
52        // 创建站内信逻辑
53        System.out.println("Creating notification for user: " + userId);
54    }
55}
56

4. 测试代码

1@SpringBootTest
2public class FanoutTest {
3    
4    @Autowired
5    private UserRegisterProducer producer;
6
7    @Test
8    public void testFanout() {
9        User user = new User();
10        user.setId(1L);
11        user.setUsername("张三");
12        user.setEmail("zhangsan@example.com");
13        user.setPhone("13800138000");
14        
15        producer.sendUserRegister(user);
16        
17        // 三个消费者都会收到消息
18    }
19}
20

3.4 路由模式（Routing）

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

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

实战案例：日志分级处理

1. 配置类

1@Configuration
2public class DirectConfig {
3    
4    /**
5     * 声明 Direct 交换机
6     */
7    @Bean
8    public DirectExchange directExchange() {
9        return new DirectExchange("log.direct.exchange");
10    }
11
12    /**
13     * 声明错误日志队列
14     */
15    @Bean
16    public Queue errorQueue() {
17        return new Queue("log.error.queue", true);
18    }
19
20    /**
21     * 声明警告日志队列
22     */
23    @Bean
24    public Queue warningQueue() {
25        return new Queue("log.warning.queue", true);
26    }
27
28    /**
29     * 声明信息日志队列
30     */
31    @Bean
32    public Queue infoQueue() {
33        return new Queue("log.info.queue", true);
34    }
35
36    /**
37     * 绑定错误日志队列
38     */
39    @Bean
40    public Binding errorBinding(Queue errorQueue, DirectExchange directExchange) {
41        return BindingBuilder.bind(errorQueue)
42            .to(directExchange)
43            .with("error");
44    }
45
46    /**
47     * 绑定警告日志队列
48     */
49    @Bean
50    public Binding warningBinding(Queue warningQueue, DirectExchange directExchange) {
51        return BindingBuilder.bind(warningQueue)
52            .to(directExchange)
53            .with("warning");
54    }
55
56    /**
57     * 绑定信息日志队列
58     */
59    @Bean
60    public Binding infoBinding(Queue infoQueue, DirectExchange directExchange) {
61        return BindingBuilder.bind(infoQueue)
62            .to(directExchange)
63            .with("info");
64    }
65}
66

2. 消息实体

1@Data
2@AllArgsConstructor
3@NoArgsConstructor
4public class LogMessage implements Serializable {
5    
6    private static final long serialVersionUID = 1L;
7    
8    private String level;      // 日志级别
9    private String message;    // 日志内容
10    private String serviceName;// 服务名称
11    private Long timestamp;    // 时间戳
12}
13

3. 生产者

1@Component
2@Slf4j
3public class LogProducer {
4    
5    @Autowired
6    private RabbitTemplate rabbitTemplate;
7
8    /**
9     * 发送错误日志
10     */
11    public void sendErrorLog(String message, String serviceName) {
12        LogMessage logMessage = new LogMessage("error", message, serviceName, System.currentTimeMillis());
13        rabbitTemplate.convertAndSend("log.direct.exchange", "error", logMessage);
14        log.info("发送错误日志: {}", message);
15    }
16
17    /**
18     * 发送警告日志
19     */
20    public void sendWarningLog(String message, String serviceName) {
21        LogMessage logMessage = new LogMessage("warning", message, serviceName, System.currentTimeMillis());
22        rabbitTemplate.convertAndSend("log.direct.exchange", "warning", logMessage);
23        log.info("发送警告日志: {}", message);
24    }
25
26    /**
27     * 发送信息日志
28     */
29    public void sendInfoLog(String message, String serviceName) {
30        LogMessage logMessage = new LogMessage("info", message, serviceName, System.currentTimeMillis());
31        rabbitTemplate.convertAndSend("log.direct.exchange", "info", logMessage);
32        log.info("发送信息日志: {}", message);
33    }
34}
35