【RabbitMQ 入门】

RabbitMQ 从入门到常用机制交换机、队列与消息怎么走适用读者后端初学者 / 准备在项目里接入消息队列的同学技术栈示例RabbitMQ 3.x · Spring AMQP概念通用示例以 Java 为主前言很多人第一次用 RabbitMQ会卡在这几个问题上消息到底发给谁是发给队列还是发给交换机Exchange、Queue、Binding、Routing Key分别干什么消费失败了怎么办ACK、NACK、重试、死信又是什么关系什么时候用 Direct什么时候用 Topic / Fanout一、先建立心智模型RabbitMQ 是一个消息中间件Broker。生产者不直接把消息塞给某个消费者而是Producer生产者 │ 发布消息 ▼ Exchange交换机──按规则路由──► Queue队列──投递──► Consumer消费者 ▲ │ Binding绑定 Routing Key路由键记住三句话生产者只认识 Exchange多数情况下不直接往队列里塞消费者只认识 Queue从队列里取消息Binding 决定「交换机里的消息如何进入哪个队列」可以把 Exchange 想成邮局分拣中心Queue 想成各个信箱Binding 是分拣规则。二、核心组件关系2.1 组件一览组件英文作用生产者Producer发送消息的应用交换机Exchange接收消息按规则转发到队列队列Queue消息的缓冲区消费者从这里取绑定BindingExchange 与 Queue 之间的路由规则路由键Routing Key发布/绑定时用的「标签」参与匹配消费者Consumer从队列消费并处理业务虚拟主机Virtual Host逻辑隔离类似命名空间权限按 vhost 划分连接/信道Connection / ChannelTCP 连接上可开多个 Channel操作都在 Channel 上完成2.2 关系图必看┌─────────────────────────────────────┐ │ Broker │ │ │ Producer ───────► │ Exchange │ │ │ │ │ │ Binding (routing key 规则) │ │ ├──────────► Queue A ──► Consumer 1 │ ├──────────► Queue B ──► Consumer 2 │ └──────────► Queue C ──► Consumer 3 │ │ └─────────────────────────────────────┘一条消息进入一个 Exchange可能进入 0 个、1 个或多个 Queue取决于交换机类型和绑定一个 Queue 可被多个 Consumer 竞争消费默认轮询分发2.3 为什么不让生产者直接写队列解耦。业务方只约定「发到哪个交换机、带什么 routing key」不必知道下游有几个队列、几个服务在消费。以后加一个审计队列只需多绑一条 Binding生产者代码不用改。三、交换机类型怎么路由RabbitMQ 常见四种 Exchange。类型决定「routing key 怎么用」。3.1 Direct直连规则队列绑定的 routing key完全相等才投递。Exchange(direct) order ├─ binding key order.pay → queue.order.pay └─ binding key order.cancel → queue.order.cancel 发布 routingKeyorder.pay → 只进 queue.order.pay适合明确的一对一/少对少路由如「支付成功通知」「取消通知」分开处理。3.2 Topic主题规则routing key 按.分段支持通配符通配符含义*匹配恰好一段#匹配零段或多段Exchange(topic) biz.topic ├─ order.* → 匹配 order.pay、order.cancel一段 ├─ order.# → 匹配 order、order.pay、order.pay.success多段 └─ *.error → 匹配 pay.error、sms.error 发布 order.pay.success → 进绑定了 order.# 的队列 → 不进 order.*因为多了一段适合日志分级、业务事件总线、一个交换机挂多种下游。3.3 Fanout广播规则忽略routing key绑定到该交换机的所有队列都收到一份拷贝。Exchange(fanout) notify.fanout ├─ → queue.sms ├─ → queue.email └─ → queue.push 发布任意消息 → 三个队列各一份适合同一事件通知多个独立系统短信、邮件、站内信。3.4 Headers较少用按消息头headers匹配而不是 routing key。灵活但不如 Topic 直观生产中相对少见。3.5 选型速查场景推荐固定几种业务类型精确投递Direct事件多、要按模式订阅Topic一发多收、完全广播Fanout需要按自定义属性过滤Headers四、队列与消息属性4.1 队列常见参数参数/特性含义durable队列元数据持久化Broker 重启后队列还在exclusive仅当前连接可见连接断开队列删除autoDelete最后一个消费者断开后自动删除TTL消息/队列超时未消费则过期可进死信max-length队列最大长度超限可丢弃或进死信惰性队列 lazy尽量落盘适合堆积场景4.2 消息持久化队列durabletrue不等于消息一定不丢消息还需deliveryMode2持久化生产端建议开启Publisher Confirm确认 Broker 已接收三者配合才能在宕机场景下尽量不丢仍非 100%极端情况要靠业务幂等。4.3 消息结构概念Envelope ├─ Exchange / Routing Key ├─ PropertiescontentType、messageId、headers、expiration… └─ Body业务 JSON / 二进制业务上通常把 JSON 放在 Body把重试次数、追踪 ID 放在 Headers。五、生产与消费的基本流程5.1 生产端伪代码概念// 1. 建立连接与信道Connectionconnectionfactory.newConnection();Channelchannelconnection.createChannel();// 2. 声明交换机、队列、绑定幂等可重复声明channel.exchangeDeclare(biz.topic,topic,true);channel.queueDeclare(queue.order.pay,true,false,false,null);channel.queueBind(queue.order.pay,biz.topic,order.pay);// 3. 发布Stringbody{\orderId\:1001,\amount\:99.00};channel.basicPublish(biz.topic,// exchangeorder.pay,// routing keyMessageProperties.PERSISTENT_TEXT_PLAIN,body.getBytes());5.2 消费端伪代码概念channel.basicQos(10);// 预取未 ACK 前最多塞给该消费者 10 条channel.basicConsume(queue.order.pay,false,(consumerTag,delivery)-{try{StringjsonnewString(delivery.getBody());handle(json);// 业务处理channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);}catch(Exceptione){// 见下文 ACK / NACKchannel.basicNack(delivery.getEnvelope().getDeliveryTag(),false,false);}},consumerTag-{});注意autoAckfalse时必须自己 Ack/Nack否则消息会一直处于 Unacked最终可能被重新投递。六、各种处理机制重点6.1 确认机制ACK / NACK / Reject操作含义basicAck处理成功Broker 删除该消息basicNack处理失败可指定是否requeuebasicReject类似 Nack但不支持批量成功 ──────────────────────────────► Ack ──► 消息从队列删除 失败 requeuetrue ───────────────► 消息重新回到队列可能立刻再被消费 失败 requeuefalse ──────────────► 丢弃若配置了死信则进入死信交换机坑对「毒消息」怎么处理都失败使用requeuetrue会形成死循环打满 CPU。生产上通常requeuefalse 死信或有限次重试后再进死信。6.2 预取 PrefetchQoSbasicQos(prefetchCount N)含义消费者最多有 N 条未确认消息。作用避免一个慢消费者被塞爆也避免公平性太差。经验CPU 型任务prefetch 可以小一点如 110快速 IO 型可适当增大6.3 竞争消费与公平分发同一队列挂多个 ConsumerQueue ──► Consumer A └─► Consumer B默认按轮询分发。谁 Ack 快谁更能继续拿新消息配合 prefetch。这是水平扩展消费者的基础加机器 加 Consumer不用改生产者。6.4 消息 TTL 与队列 TTL消息 TTL单条消息的过期时间队列 TTL队列里所有消息的统一过期时间过期后消息被丢弃若队列配置了死信交换机则进入死信链路。常用于订单超时关闭的「延迟效果」配合死信见样例 3。6.5 死信DLX / DLQ死信交换机Dead Letter Exchange不是特殊类型只是一个普通 Exchange被指定为「死信去处」。消息变成死信的常见原因消费者Nack/Reject且requeuefalse消息 TTL 过期队列达到最大长度消息被挤出业务队列 order.queue 参数 x-dead-letter-exchange order.dlx x-dead-letter-routing-key order.dead 失败 / 过期 / 超长 │ ▼ order.dlx ──► order.dead.queue死信队列 │ ▼ 对账 / 告警 / 人工处理死信的价值失败消息与正常流量隔离主队列继续跑失败可事后补偿。6.6 幂等消费MQ至少一次投递很常见网络重试、Consumer 重启未 Ack 等所以业务必须幂等同一 messageId / 业务唯一键 处理两次 只产生一次副作用常见手段数据库唯一索引订单号、幂等键RedisSETNX处理标记状态机只允许CREATED → PAID重复支付直接返回成功没有幂等的重试/死信都可能把账做花。6.7 发布确认与返回生产端可靠机制作用Publisher ConfirmBroker 确认已收到或写入磁盘策略满足Mandatory Return消息无法路由到任何队列时退回生产者适合对「发出去」有强要求的场景和消费端 Ack 是两回事。6.8 事务 vs ConfirmChannel 事务能保证发布原子性但性能差。现代实践更推荐Publisher Confirm而不是 AMQP 事务。七、样例样例 1Direct —— 订单支付结果通知需求支付成功、支付失败走不同处理逻辑。Exchange: order.direct (direct) order.pay.success → queue.pay.success order.pay.fail → queue.pay.fail发布channel.basicPublish(order.direct,order.pay.success,null,body);消费成功队列的服务发积分失败队列的服务发站内信。互不影响。样例 2Topic —— 日志收集需求所有*.error进告警队列order.#进订单分析队列。Exchange: log.topic (topic) *.error → queue.alert order.# → queue.order.analytics发布 routingKey order.pay.error → 同时进入 queue.alert 和 queue.order.analytics各一份拷贝这就是 Topic 的典型用法一条事件多种订阅。样例 3TTL 死信 ——「延迟 30 分钟」关单简化版RabbitMQ 本身没有「延迟队列」一等公民可用插件经典做法是1. 消息发到 delay.queue设置 TTL 30min 2. delay.queue 不挂业务消费者 3. 过期后经 DLX 进入 close.order.queue 4. close.order.queue 的消费者执行「若仍未支付则关单」Producer ──► delay.exchange ──► delay.queue (TTL30m, DLXclose.exchange) │ 到期 ▼ close.exchange ──► close.order.queue ──► Consumer注意队列级 TTL 时过期顺序与队列头有关高精度延迟更推荐延迟插件或时间轮方案。样例 4Fanout —— 用户注册成功后的多通道通知Exchange: user.registered (fanout) → queue.mail → queue.sms → queue.growth成长体系发新人礼注册服务只发一条「用户已注册」事件三个下游独立扩缩容。某个通知挂了不影响另外两个各自队列堆积。样例 5消费失败进死信 对账推荐心智主队列 work.queue 成功 → Ack 失败 → Nack(requeuefalse) → DLQ 死信队列 work.dlq → 记录日志 / 告警 → 根据业务键做补偿回滚库存、标记失败单等 → 一般不要无限重投主队列伪代码voidonMessage(Messagemsg,Channelchannel,longtag)throwsIOException{try{process(msg);// 业务内部保证幂等channel.basicAck(tag,false);}catch(Exceptione){log.error(process failed,e);channel.basicNack(tag,false,false);// 进死信不 requeue}}voidonDeadLetter(Messagemsg,Channelchannel,longtag)throwsIOException{try{reconcile(msg);// 对账补偿而不是盲目重试channel.basicAck(tag,false);}catch(Exceptione){log.error(reconcile failed, need manual,e);channel.basicNack(tag,false,false);// 避免死信里死循环}}八、Spring AMQP 声明拓扑示意概念与原生客户端一致只是用 Bean 声明BeanDirectExchangeorderExchange(){returnnewDirectExchange(order.direct,true,false);}BeanQueuepaySuccessQueue(){returnQueueBuilder.durable(queue.pay.success).withArgument(x-dead-letter-exchange,order.dlx).withArgument(x-dead-letter-routing-key,order.dead).build();}BeanBindingpaySuccessBinding(QueuepaySuccessQueue,DirectExchangeorderExchange){returnBindingBuilder.bind(paySuccessQueue).to(orderExchange).with(order.pay.success);}消费RabbitListener(queuesqueue.pay.success)publicvoidonPaySuccess(Stringbody,Channelchannel,Header(AmqpHeaders.DELIVERY_TAG)longtag)throwsIOException{try{// ...channel.basicAck(tag,false);}catch(Exceptione){channel.basicNack(tag,false,false);}}application.yml中常见配置spring:rabbitmq:host:localhostport:5672username:guestpassword:guestvirtual-host:/publisher-confirm-type:correlatedlistener:simple:acknowledge-mode:manualprefetch:10管理台默认http://localhost:15672用户 guest/guest仅限本地。九、常见误区误区正解消息发给队列通常发给Exchange由 Binding 进队列队列持久化 消息不丢还要消息持久化 Confirm 等失败就 requeuetrue毒消息会死循环优先死信或有限重试有了 MQ 就不需要幂等MQ 常至少一次必须幂等Fanout 还要精心设计 routing keyFanout忽略routing key一个业务一个 Broker 连接狂开复用 Connection多开 Channel十、一张图串起来Producer │ basicPublish(exchange, routingKey, body) ▼ Exchange ──Binding──► Queue ──basicDeliver──► Consumer │ ├─ Ack ──────────► 删除消息 ├─ Nackrequeue ► 回队列 └─ Nack 无 requeue DLX ► 死信队列 ► 补偿/告警十一、学习路径建议本地起 RabbitMQ打开 15672手动创建 Direct 队列用管理台发一条消息写最小 Producer / Consumer观察 Ack 前后 Unacked 数量变化故意消费失败对比requeuetrue/false给队列挂上 DLX看消息如何进死信队列再用 Topic / Fanout 各做一个小样例当你能不看文档画出「交换机 → 绑定 → 队列 → 消费者 → 死信」时接入任何业务项目都会轻松很多。十二、小结Exchange 负责路由Queue 负责存储与投递Binding Routing Key 决定路径Direct 精确、Topic 模式、Fanout 广播—— 按场景选型消费端用手动 Ack失败慎用无限 requeue死信用于隔离与补偿生产端 Confirm、消息持久化、消费端幂等—— 可靠投递的三件套消息队列解决的是异步、解耦、削峰它不自动保证业务正确。业务正确仍然靠清晰的状态机、幂等、以及失败后的补偿路径。