RabbitMQ 的特点是路由灵活：消息不直接进队列，而是先到 Exchange，再由绑定规则分发。这让它在「一对多、按规则分发」的内部系统间解耦里很好用。
一、AMQP 模型
二、四种 Exchange 选型
| 类型 | 路由规则 | 典型场景 |
|---|---|---|
direct |
routing key 完全相等 | 点对点任务队列（最常用） |
fanout |
忽略 key，广播到所有绑定队列 | 一个事件触发多个下游（如订单创建 → 扣库存、发券、通知） |
topic |
通配符：* 一个词，#. 多个词 |
按业务维度订阅（order.#、*.pay） |
headers |
按消息头属性匹配 | 很少用，性能差 |
声明示例（Spring AMQP）：
@Bean
public TopicExchange orderExchange() {
return new TopicExchange("order.exchange", true, false); // durable=true, autoDelete=false
}
@Bean
public Queue payQueue() {
return QueueBuilder.durable("order.pay.queue")
.deadLetterExchange("order.dlx") // 死信交换机
.deadLetterRoutingKey("order.pay.dead")
.ttl(60000) // 队列级 TTL（可选）
.maxLength(100000) // 队列最大长度，防堆积写满磁盘
.build();
}
@Bean
public Binding payBinding() {
return BindingBuilder.bind(payQueue()).to(orderExchange()).with("order.pay");
}
三个持久化开关必须都开，否则重启消息就没了：Exchange durable=true、Queue durable=true、消息 deliveryMode=PERSISTENT。
三、可靠投递
spring:
rabbitmq:
publisher-confirm-type: correlated # 生产者确认：消息是否到达 Exchange
publisher-returns: true # 消息无法路由时回调
listener:
simple:
acknowledge-mode: manual # ★ 手动 ack
prefetch: 50 # ★ 每次只推 50 条，防止消费者被压垮
retry:
enabled: true
max-attempts: 3
Confirm 回调（生产端）：
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
if (!ack) {
// 落库 + 定时任务重发，别在这里死循环重试
msgFailService.record(correlationData.getId(), cause);
}
});
手动 ack（消费端）：
@RabbitListener(queues = "order.pay.queue")
public void handle(Message message, Channel channel) throws IOException {
long tag = message.getMessageProperties().getDeliveryTag();
try {
bizService.handle(message);
channel.basicAck(tag, false); // 单条确认
} catch (BizException e) {
log.warn("业务失败不重试", e);
channel.basicAck(tag, false); // ★ 确认掉，避免无限重试
deadLetterService.save(message); // 自己落死信表
} catch (Exception e) {
log.error("可重试异常", e);
channel.basicNack(tag, false, true); // requeue=true，重新入队
}
}
basicNack(tag, false, true) 无限重试会把 MQ 拖死。正确做法是限制重试次数（用 Redis 计数），超过就转死信队列。
四、死信队列 DLX
消息变成死信的三种情况：
- 被
basic.reject/basic.nack且requeue=false - 消息 TTL 过期
- 队列达到最大长度被丢弃
// 死信交换机和队列（都是普通交换机/队列，只是被指定为 DLX）
@Bean
public DirectExchange dlx() { return new DirectExchange("order.dlx", true, false); }
@Bean
public Queue dlq() { return QueueBuilder.durable("order.dlq").build(); }
@Bean
public Binding dlBinding() {
return BindingBuilder.bind(dlq()).to(dlx()).with("order.pay.dead");
}
死信队列必须有人消费或至少有人看。我一般配一个独立的消费者把死信落表 + 告警，并提供「人工重试」入口。
五、延迟消息的两种实现
方案一：TTL + DLX（原生，推荐）
给消息设 expiration（毫秒），发到一个没有消费者的队列，TTL 到期后自动成为死信，转发到真正的业务队列：
rabbitTemplate.convertAndSend("delay.exchange", "delay.key", msg, m -> {
m.getMessageProperties().setExpiration("30000"); // 30 秒后投递
return m;
});
坑：RabbitMQ 的 TTL 检查是惰性的 —— 只检查队首消息是否过期。如果队首是 10 分钟的消息，后面一条 10 秒的会被堵住。所以同一队列不要混用不同 TTL，按延迟档位建多个队列（如 5s/30s/5min/1h）。
方案二：延迟插件 rabbitmq_delayed_message_exchange（推荐用于任意延迟）
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
// 交换机类型用 x-delayed-message
@Bean
public CustomExchange delayExchange() {
Map<String, Object> args = new HashMap<>();
args.put("x-delayed-type", "direct");
return new CustomExchange("delay.ex", "x-delayed-message", true, false, args);
}
// 发送时带 x-delay 头（毫秒）
m.getMessageProperties().setHeader("x-delay", 60000);
插件方案支持任意延迟且不受队首阻塞影响，但延迟消息存在 Mnesia 表里，节点重启会丢失，且集群下不支持镜像（对可靠性要求高的场景要评估）。
六、常用命令
# 队列与消息数
rabbitmqctl list_queues name messages messages_ready messages_unacknowledged
# 连接与通道
rabbitmqctl list_connections
rabbitmqctl list_channels
# 消费者
rabbitmqctl list_consumers
# 集群状态
rabbitmqctl cluster_status
# 启用管理插件（Web 控制台 15672）
rabbitmq-plugins enable rabbitmq_management
# 内存与磁盘水位（默认磁盘低于 50MB 会阻塞生产者！）
rabbitmqctl status | grep -A5 disk_free_limit
# 设置内存水位（超过就阻塞生产者，防 OOM）
rabbitmqctl set_vm_memory_high_watermark 0.6
七、一次堆积把磁盘写满的事故
脱敏自某 SaaS 商城的订单异步下单链路。
现象：RabbitMQ 磁盘告警，生产者全部阻塞，下单接口大面积超时。
排查：
rabbitmqctl list_queues name messages # 发现 order.create.queue 堆了 800 万条
rabbitmqctl list_consumers # 消费者还在，但 unacked 数量巨大
unacked 巨大说明消费者拿到了消息但一直没 ack —— jstack 看消费端，线程全卡在调用支付回调接口上（下游超时 30s），而 prefetch 没设（默认无限推送），RabbitMQ 把几百万条消息全推给了消费者，消费者内存也快撑爆了。
处理：
- 先把
prefetch设成 50（要重启消费者），限制同时处理的消息数。 - 支付回调超时从 30s 降到 3s，失败进重试队列。
- 给队列加
x-max-length，并设置x-overflow: reject-publish（满了就拒绝生产者，而不是无限堆积写满磁盘）。 - 开启磁盘/内存水位告警，低于阈值提前扩容。
结论：RabbitMQ 的默认配置偏向「尽力投递」，生产上必须显式限制 prefetch、队列长度和磁盘水位，否则一个慢消费者就能拖垮整个 Broker。
八、高可用
| 方案 | 说明 | 建议 |
|---|---|---|
| 普通集群 | 队列只在一个节点，元数据共享 | 不保证高可用，节点挂了队列不可用 |
| 镜像队列（Classic Queue Mirroring） | 队列复制到多节点 | 老方案，RabbitMQ 3.x 仍可用 |
| Quorum Queue | 基于 Raft 的复制队列 | 新项目首选，数据安全性更好，但延迟略高、不支持 TTL 之外的一些特性 |
生产建议：新队列用 Quorum Queue，并对磁盘做监控。RabbitMQ 在磁盘剩余空间低于 disk_free_limit（默认 50MB）时会阻塞所有生产者，这是它最危险的默认行为之一，务必配告警。
一起聊聊