RabbitMQ

RabbitMQ 学习笔记

消息队列就像快递小哥，帮你把消息从生产者送到消费者手中！
——某位刚被同步调用折磨完的程序员

一、RabbitMQ的介绍：什么是消息队列？

1.1 什么是MQ？（消息队列）

MQ（Message Queue，消息队列）：消息中间件，用于微服务之间进行通信。

生产者（Sender） ——–> 消息队列（MQ） ——–> 消费者（Consumer）

💡 形象比喻：
生产者 = 发快递的人
消息队列 = 快递站
消费者 = 收快递的人
消息 = 快递包裹

1.2 什么是AMQP？（高级消息队列协议）

AMQP（Advanced Message Queuing Protocol）：高级消息队列协议，RabbitMQ的实现标准。

AMQP的核心组件：

🔥 重要：Channel是轻量级的，一个Connection可以创建多个Channel，避免频繁建立连接的开销！

1.3 什么是RabbitMQ？

RabbitMQ 是用 Erlang语言 编写的实现AMQP协议的消息中间件。

💡 为什么用Erlang？
Erlang天生适合高并发、分布式系统，就像"为消息队列而生"！

1.4 为什么要使用MQ？（三大好处）

1️⃣ 解耦 – 让系统不再"牵一发而动全身"

场景：A服务本来只调用B、C服务，若添加D服务…

• 不用MQ：修改A服务代码，重新部署，测试…
• 用MQ：A服务只管发消息，B、C、D谁想消费谁消费，互不干扰！

🎯 好处：系统扩展性大幅提升，修改一个服务不影响其他服务！

2️⃣ 异步 – 让系统飞起来

场景：用户下单后需要发短信、发邮件、更新库存…

• 同步调用：用户要等所有操作完成才能看到"下单成功"（慢！）
• 异步调用：用户下单后立即返回成功，后台慢慢处理其他操作（快！）

用户下单
↓
发消息到MQ
↓
立即返回"下单成功" ← 用户看到结果
↓
消费者慢慢处理：发短信、发邮件、更新库存…

🎯 好处：响应速度提升，用户体验更好！

3️⃣ 削峰 – 应对流量洪峰

场景：秒杀活动，瞬间10万请求涌入…

• 不用MQ：服务器直接宕机！
• 用MQ：请求先存到队列，消费者按处理能力慢慢拉取

10万请求 → 消息队列（缓冲） → 消费者（每秒处理1000条）

🎯 好处：系统稳定性提升，不会被瞬间流量冲垮！

1.5 RabbitMQ的启动器

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

二、RabbitMQ的安装（Linux版）

💡 前提：需要先安装Erlang环境！

2.1 安装Erlang语言环境

cd /usr/upload
rpm -ivh esl-erlang-17.3-1.x86_64.rpm –force –nodeps
rpm -ivh esl-erlang_17.3-1~centos~6_amd64.rpm –force –nodeps
rpm -ivh esl-erlang-compat-R14B-1.el6.noarch.rpm –force –nodeps

⚠️ 注意：–force –nodeps 表示强制安装，忽略依赖关系（生产环境慎用！）

2.2 安装RabbitMQ

rpm -ivh rabbitmq-server-3.4.1-1.noarch.rpm

2.3 启动和关闭

# 启动
service rabbitmq-server start

# 查看状态
service rabbitmq-server status

# 设置开机自启
chkconfig rabbitmq-server on

# 停止
service rabbitmq-server stop

# 重启
service rabbitmq-server restart

2.4 安装后台管理界面

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

# 重启服务
service rabbitmq-server restart

🎯 好处：可以通过Web界面管理RabbitMQ，可视化操作超方便！

2.5 创建账户

# 添加用户
rabbitmqctl add_user admin 1111

# 设置用户角色（administrator表示管理员）
rabbitmqctl set_user_tags admin administrator

# 设置权限（.* 表示所有权限）
rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"

# 查看用户列表
rabbitmqctl list_users

2.6 测试访问

打开浏览器访问：
👉 http://192.168.61.137:15672

• 用户名：admin
• 密码：1111

看到管理界面？恭喜！RabbitMQ安装成功！

三、五种消息模型（核心！）

3.1 Simple消息模型（点对点）

生产者 ——–> 队列 ——–> 消费者

特点：一个生产者，一个队列，一个消费者

手动ACK（确认）机制

channel.basicConsume(queueName, false, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
try {
// 处理消息…
System.out.println("收到消息: " + new String(body));

// 手动确认（告诉MQ：消息已成功消费）
channel.basicAck(envelope.getDeliveryTag(), false);
} catch (Exception e) {
// 处理失败，拒绝消息（可以选择重新入队）
channel.basicNack(envelope.getDeliveryTag(), false, true);
}
}
});

🔥 问题：如何保证消费者把消息成功消费？
答案：手动ACK！只有消费者处理成功后才确认，失败则拒绝或重新入队！

3.2 Work消息模型（任务队列）

生产者 ——–> 队列 ——–> 消费者1
|
——–> 消费者2
|
——–> 消费者3

特点：一个生产者，一个队列，多个消费者（负载均衡）

能者多劳（公平分发）

// 设置预取数量为1（消费者处理完当前消息后才接收下一条）
channel.basicQos(1);

channel.basicConsume(queueName, false, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
try {
// 模拟处理时间
Thread.sleep(1000);
System.out.println("消费者1处理: " + new String(body));
channel.basicAck(envelope.getDeliveryTag(), false);
} catch (Exception e) {
channel.basicNack(envelope.getDeliveryTag(), false, true);
}
}
});

🔥 问题：如何防止消息堆积？
答案：多个消费者 + 能者多劳（basicQos）！

3.3 Fanout消息模型（广播）

生产者 ——–> 交换机 ——–> 队列1 ——–> 消费者1
|
——–> 队列2 ——–> 消费者2
|
——–> 队列3 ——–> 消费者3

特点：一个生产者，一个Fanout交换机，多个队列，每个队列绑定一个消费者

代码示例：

// 声明Fanout交换机
channel.exchangeDeclare("fanout_exchange", "fanout");

// 声明队列
String queueName1 = channel.queueDeclare().getQueue();
String queueName2 = channel.queueDeclare().getQueue();

// 绑定队列到交换机（Fanout不需要routingKey）
channel.queueBind(queueName1, "fanout_exchange", "");
channel.queueBind(queueName2, "fanout_exchange", "");

// 发送消息
channel.basicPublish("fanout_exchange", "", null, "广播消息".getBytes());

⚠️ 注意：Exchange只负责分发消息，若没有队列绑定到Exchange上，消息会被丢弃！

3.4 Direct消息模型（路由）

生产者 ——–> 交换机 ——–> 队列1（routingKey=info） ——–> 消费者1
|
——–> 队列2（routingKey=error） ——–> 消费者2

特点：通过routingKey精确控制消息分发

代码示例：

// 声明Direct交换机
channel.exchangeDeclare("direct_exchange", "direct");

// 声明队列
String infoQueue = channel.queueDeclare().getQueue();
String errorQueue = channel.queueDeclare().getQueue();

// 绑定队列到交换机（指定routingKey）
channel.queueBind(infoQueue, "direct_exchange", "info");
channel.queueBind(errorQueue, "direct_exchange", "error");

// 发送消息（指定routingKey）
channel.basicPublish("direct_exchange", "info", null, "信息日志".getBytes());
channel.basicPublish("direct_exchange", "error", null, "错误日志".getBytes());

🎯 routingKey：就像快递单上的"收件地址"，Exchange根据它决定消息发到哪个队列！

3.5 Topic消息模型（主题/通配符）

生产者 ——–> 交换机 ——–> 队列1（routingKey=*.info.#） ——–> 消费者1
|
——–> 队列2（routingKey=order.*） ——–> 消费者2

特点：支持通配符匹配，更灵活的路由规则

通配符规则：

代码示例：

// 声明Topic交换机
channel.exchangeDeclare("topic_exchange", "topic");

// 声明队列
String allInfoQueue = channel.queueDeclare().getQueue();
String orderQueue = channel.queueDeclare().getQueue();

// 绑定队列（使用通配符）
channel.queueBind(allInfoQueue, "topic_exchange", "*.info.#");
channel.queueBind(orderQueue, "topic_exchange", "order.*");

// 发送消息
channel.basicPublish("topic_exchange", "user.info.login", null, "用户登录".getBytes());
channel.basicPublish("topic_exchange", "order.create", null, "创建订单".getBytes());

🎯 应用场景：日志系统（info、warn、error分级）、订单系统（不同业务类型）

四、持久化（保证消息不丢失）

4.1 Exchange持久化

// 第三个参数true表示持久化
channel.exchangeDeclare("my_exchange", "direct", true);

4.2 Queue持久化

// 第二个参数true表示持久化
channel.queueDeclare("my_queue", true, false, false, null);

4.3 消息持久化

// 使用MessageProperties.PERSISTENT_TEXT_PLAIN标记消息为持久化
channel.basicPublish("my_exchange", "my_routingKey",
MessageProperties.PERSISTENT_TEXT_PLAIN,
"持久化消息".getBytes());

4.4 完整的持久化示例

// 1. 声明持久化的Exchange
channel.exchangeDeclare("persistent_exchange", "direct", true);

// 2. 声明持久化的Queue
channel.queueDeclare("persistent_queue", true, false, false, null);

// 3. 绑定
channel.queueBind("persistent_queue", "persistent_exchange", "persistent_key");

// 4. 发送持久化消息
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
.deliveryMode(2) // 2表示持久化
.contentType("text/plain")
.build();

channel.basicPublish("persistent_exchange", "persistent_key", props, "重要消息".getBytes());

🔥 问题：如何保证消息不丢失？
答案：三重保险！
1️⃣ Exchange持久化
2️⃣ Queue持久化
3️⃣ 消息持久化

五、SpringBoot整合RabbitMQ

5.1 添加依赖（pom.xml）

<dependencies>
<!– RabbitMQ启动器 –>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

<!– 测试依赖 –>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>

5.2 配置文件（application.yml）

spring:
rabbitmq:
host: 192.168.61.137 # RabbitMQ服务器地址
port: 5672 # 端口
username: admin # 用户名
password: 1111 # 密码
virtual-host: / # 虚拟主机

# 消息确认配置
publisher-confirm-type: correlated # 开启发布确认
publisher-returns: true # 开启失败回调

# 消费者配置
listener:
simple:
acknowledge-mode: manual # 手动ACK
concurrency: 5 # 最小消费者数量
max-concurrency: 10 # 最大消费者数量
prefetch: 1 # 每次预取1条消息（能者多劳）

5.3 消费者（Receiver）

@Component
public class Receiver {

/**
* 接收消息的三个要素：
* 1、exchange（交换机）
* 2、queue（队列）
* 3、routingKey（路由键）
*/
@RabbitListener(bindings = @QueueBinding(
value = @Queue(value = "springboot_queue", durable = "true"), // 持久化队列
exchange = @Exchange(value = "springboot_exchange", type = ExchangeTypes.TOPIC, durable = "true"), // 持久化交换机
key = {"springboot.*"} // 路由键（支持通配符）
))
public void listenerMsg(String msg, Channel channel, Message message) {
try {
System.out.println("【Receiver】收到消息: " + msg);

// 业务处理…
// Thread.sleep(1000);

// 手动ACK
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
System.out.println("【Receiver】消息确认成功");
} catch (Exception e) {
e.printStackTrace();
try {
// 拒绝消息，重新入队
channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
System.out.println("【Receiver】消息处理失败，重新入队");
} catch (IOException ioException) {
ioException.printStackTrace();
}
}
}
}

5.4 生产者（Sender）

@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest(classes = {SpringbootRabbitmqApp.class})
public class Sender {

@Autowired
private AmqpTemplate amqpTemplate;

@Test
public void testSendMsg() {
String msg = "师姐你好-2";

// 发送消息：交换机、路由键、消息内容
amqpTemplate.convertAndSend("springboot_exchange", "springboot.test", msg);

System.out.println("【Sender】发送消息: " + msg);

try {
Thread.sleep(100000); // 等待消费者处理
} catch (InterruptedException e) {
e.printStackTrace();
}
}

// 批量发送消息
@Test
public void testBatchSend() {
for (int i = 1; i <= 10; i++) {
String msg = "批量消息-" + i;
amqpTemplate.convertAndSend("springboot_exchange", "springboot.batch", msg);
System.out.println("发送: " + msg);
}
}
}

5.5 高级配置：消息确认回调

@Configuration
public class RabbitMQConfig {

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

// 消息发送成功回调
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
if (ack) {
System.out.println("【Confirm】消息发送成功: " + correlationData);
} else {
System.out.println("【Confirm】消息发送失败: " + cause);
}
});

// 消息发送失败回调（路由失败）
rabbitTemplate.setReturnsCallback(returnedMessage -> {
System.out.println("【Return】消息路由失败: " + returnedMessage.getMessage());
});

return rabbitTemplate;
}
}

5.6 死信队列（DLQ）配置

@Configuration
public class DeadLetterQueueConfig {

// 正常队列
@Bean
public Queue normalQueue() {
Map<String, Object> args = new HashMap<>();
// 设置死信交换机
args.put("x-dead-letter-exchange", "dlx_exchange");
// 设置死信路由键
args.put("x-dead-letter-routing-key", "dlx_key");
// 设置消息过期时间（毫秒）
args.put("x-message-ttl", 10000);

return new Queue("normal_queue", true, false, false, args);
}

// 死信交换机
@Bean
public DirectExchange dlxExchange() {
return new DirectExchange("dlx_exchange", true, false);
}

// 死信队列
@Bean
public Queue deadLetterQueue() {
return new Queue("dead_letter_queue", true);
}

// 绑定死信队列到死信交换机
@Bean
public Binding dlxBinding() {
return BindingBuilder.bind(deadLetterQueue())
.to(dlxExchange())
.with("dlx_key");
}
}

💡 死信队列应用场景：

六、常见问题与解决方案

6.1 消息丢失问题

6.2 消息重复消费

原因：消费者处理成功但ACK失败，消息重新入队

解决方案：

// 幂等性示例
@Transactional
public void processOrder(String orderId) {
// 检查订单是否已处理
if (orderService.isProcessed(orderId)) {
return; // 已处理，直接返回
}

// 处理订单…
orderService.createOrder(orderId);

// 标记为已处理
orderService.markAsProcessed(orderId);
}

6.3 消息堆积

原因：生产者速度 > 消费者速度

解决方案：

// 惰性队列配置
@Bean
public Queue lazyQueue() {
Map<String, Object> args = new HashMap<>();
args.put("x-queue-mode", "lazy"); // 惰性队列
return new Queue("lazy_queue", true, false, false, args);
}

七、总结：RabbitMQ使用Checklist

✅ 基础配置：安装Erlang + RabbitMQ + 管理插件
✅ 安全设置：创建专用账户，设置权限
✅ 消息模型：根据场景选择合适模型（Simple/Work/Fanout/Direct/Topic）
✅ 持久化：Exchange、Queue、消息三重持久化
✅ 手动ACK：消费者处理成功后再确认
✅ 死信队列：处理失败消息，避免丢失
✅ 监控告警：配置队列长度、消费速率监控
✅ 幂等性：业务逻辑支持重复消费