从“零”到“高可用集群 + 黑科技”一步一图一命令，所有代码基于 Spring Boot 3.2 + RabbitMQ 3.12 实测可拷走。
阶段 0：学前 3 问——RabbitMQ 能干嘛？
|
异步解耦 |
订单创建后 50 ms 内返回用户，后续库存、积分、短信全走消息 |
|
削峰填谷 |
秒杀 5w QPS 先落队列，后端按 1k QPS 消费，DB 不炸 |
|
可靠传输 |
生产者、队列、消费者三重确认，金融级不丢消息 |
阶段 1：安装——Win/Mac/Linux 3 行命令
1.装 Erlang（RabbitMQ 用它写）
# Mac (Homebrew)
brew install erlang
# Ubuntu
sudo apt install erlang-base erlang-dev
2.装 RabbitMQ
# Mac
brew install rabbitmq
# Ubuntu
sudo apt install rabbitmq-server
3.启管理插件（Web UI 15672）
rabbitmq-plugins enable rabbitmq_management
sudo systemctl start rabbitmq-server
浏览器访问 http://localhost:15672 默认 guest/guest 登录成功即 OK 。
阶段 2：最简配置——Spring Boot 3 一 yml 搞定
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
virtual-host: / # 默认
vhost publisher-confirm-type: correlated # 生产确认
publisher-returns: true # 失败返回
listener:
simple:
acknowledge-mode: manual # 手动
ack concurrency: 5 # 并发 5
max-concurrency: 10
依赖：
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
阶段 3：核心模型——一张图记住 6 大概念
Producer → Exchange → RoutingKey → Queue → Consumer
↑
BindingKey
-
Exchange 类型 = 路由算法：direct（精准）| topic（模式）| fanout（广播）| headers（少用）
-
Queue 才是真正存消息的容器；Exchange 只负责转发
-
Channel 是 TCP 内的轻量连接，线程安全，推荐一条线程一个 Channel
阶段 4：Hello World——生产&消费 10 行代码
配置类（自动声明队列）：
@Configuration
public class SimpleConfig {
@Bean
public Queue helloQueue() {
return QueueBuilder.durable("hello.q").build();
}
}
生产者：
@RestController
class Producer {
@Autowired private RabbitTemplate template;
@GetMapping("/send")
public String send(@RequestParam String msg) {
template.convertAndSend("hello.q", msg);
return "ok";
}
}
消费者：
@Component
public class HelloConsumer {
@RabbitListener(queues = "hello.q")
public void receive(String msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
System.out.println("收到 = " + msg);
channel.basicAck(tag, false); // 手动 ack
}
}
启动后访问 http://localhost:8080/send?msg=abc 控制台打印即成功 。
阶段 5：4 种交换机——口诀一张表
|
direct |
完全匹配 RoutingKey |
info.log → info.log |
单播、任务分发 |
|
topic |
模式匹配 * 单层 # 多层 |
order.* |
分类订阅 |
|
fanout |
无视 key，广播全部队列 |
无 |
公告、配置下发 |
|
headers |
头部 KV 匹配 |
x-match=any |
少用大流量 |
代码示例（topic 模式）：
@Bean
public TopicExchange topicEx() {
return ExchangeBuilder.topicExchange("topic.ex").durable(true).build();
}
@Bean
public Binding topicBinding() {
return BindingBuilder
.bind(QueueBuilder.durable("topic.q").build())
.to(topicEx())
.with("order.*");
}
阶段 6：消息可靠性——3 确认 2 补偿
1.生产者确认（callback）
template.setConfirmCallback((correlation, ack, cause) -> {
if (ack) log.info("消息已送达交换机");
else log.error("消息未送达，cause={}", cause);
});
2.mandatory=true + 返回回调 → 交换机→队列失败时触发
template.setReturnsCallback(returned -> log.error("队列投递失败，消息={}",
returned.getMessage()));
3.消费者手动 ack / nack / reject
channel.basicAck(tag, false); // 成功
channel.basicNack(tag, false, true); // 失败重入队
channel.basicReject(tag, false); // 抛弃/死信
面试口诀：“投递确认 + 失败返回 + 消费ack” 3 连击，RabbitMQ 不丢消息 。
阶段 7：死信队列（DLX）——失败消息归宿
死信触发条件：拒绝且 requeue=false / 超时 / 队列满 配置：
@Bean
public Queue bizQueue() {
return QueueBuilder.durable("biz.q")
.deadLetterExchange("dlx.ex")
.deadLetterRoutingKey("dlx") .ttl(5000) // 5 秒超时
.build();
}
@Bean
public Queue dlxQueue() {
return QueueBuilder.durable("dlx.q").build();
}
@Bean
public DirectExchange dlxEx() {
return ExchangeBuilder.directExchange("dlx.ex").build();
}
@Bean
public Binding dlxBind() {
return BindingBuilder.bind(dlxQueue()).to(dlxEx()).with("dlx");
}
消费 dlx.q 即可统一处理超时/失败订单，做补偿或告警。
阶段 8：延迟消息——2 种实现
TTL + DLX（无插件，精度秒级） 见阶段 7 代码，队列级 TTL 或消息级 TTL均可。
插件 rabbitmq_delayed_message_exchange（毫秒级）
安装：
rabbitmq-plugins enable rabbitmq_delayed_message_exchangeJava
代码：
@Bean
public CustomExchange delayEx() {
return ExchangeBuilder
.customExchange("delay.ex", "x-delayed-message")
.args(Map.of("x-delayed-type", "direct"))
.build();
}
发送时加头：
template.convertAndSend("delay.ex", "delay.key", msg, m -> {
m.getMessageProperties().setDelay(3000); // 3 秒
return m;
});
阶段 9：批量与异步——高吞吐玩法
异步发送（不阻塞主线程）：
template.setUsePublisherConnection(true); // 独立连接
CompletableFuture<Void> future = template.asyncSend("batch.q", msg);
批量发送（减少 RTT）：
List<Message> batch = msgs.stream()
.map(m ->MessageBuilder.withBody(m.getBytes()).build())
.collect(Collectors.toList());
template.invoke(t -> {
batch.forEach(b -> t.convertAndSend("batch.q", b));
return null;
}, null);
阶段10：监控 & 指标——一眼看出瓶颈
1.管理 UI http://host:15672 → Queues 标签看 Message rates / Deliver / Ack
2.Prometheus + Grafana 启用插件：
rabbitmq-plugins enable rabbitmq_prometheus
端口 15692 暴露指标，Grafana 模板 RabbitMQ-Overview 直接导入，Delivery rate、Memory、Disk space 三红线预警。
3.消费者 Lag = Ready 消息数，持续 > 1w 即要扩容或加线程。
阶段11：集群 & 高可用——3 节点最小
普通集群（队列仍单点）：
# node2 加入 node1
rabbitmqctl stop_app
rabbitmqctl join_cluster
rabbit@node1 rabbitmqctl start_app
镜像队列（高可用）：
rabbitmqctl set_policy ha "^" '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'
镜像数量 = 2 即可，=节点数 会全网复制，IO 爆炸 。
故障切换： 节点宕机 → 镜像队列自动提升为 master，业务无感知；原节点恢复 → 自动同步后再次加入集群。
阶段12：黑科技 & 踩坑大全
黑科技
流式插件 rabbitmq-stream（Kafka 风格） 百万级 QPS、分段日志，一条连接可并发读，适合日志/埋点。
Federation = 跨机房消息复制，异地多活利器。
Shovel = 动态迁移队列，0 停机搬迁数据。
MQTT 插件 → 物联网设备直接 PUBLISH 到 RabbitMQ。
内部 tracing → 启用 rabbitmq_tracing，消息路径全链路落盘，排障神器。
踩坑 & 急救
|
连接泄漏 |
connection_count 飙高 |
一定 channel.close()；Spring 默认缓存，别自己 new |
|
消息堆积 |
Ready 数暴涨 |
加消费者线程 or 扩容队列；禁止 autoAck=true |
|
磁盘写满 |
阻塞所有 publish |
设 disk_free_limit.absolute=2GB；日志/镜像及时清理 |
|
镜像过多 |
性能暴跌 |
镜像数 ≤ 2，≠ 节点数 |
|
TTL 无效 |
延迟不生效 |
必须队列级或消息级二选一，同时存在以最小值为准 |
总结口诀（背诵 30 秒）
“Exchange 管路由，Queue 才存货； 3 确认 2 死信，集群镜像别过多； TTL+DLX 做延迟，Stream 插件破百万； 连接要关，镜像要清，Prometheus 看 Lag！”
收藏这篇，从单机到异地多活，RabbitMQ 任你摆布！
