尧图网页设计 
做后端开发这几年RabbitMQ 算是我最常用也最常被问到的一个消息中间件。网上资料不少但很多朋友卡在第一步五大消息模型到底是什么为什么有的模型需要交换机有的模型只靠队列就能跑通这篇文章就是把基础彻底补上——按官方 Tutorial 的脉络把简单队列、工作队列、发布/订阅、路由、主题这五种模型拆开讲一遍代码全部带详细注释顺手把 Windows 和 Linux 下安装、启动失败、改端口这些实战坑也一起填了。适合刚接触 RabbitMQ 的读者也适合已经用过一段时间、但一直没把模型关系真正理清的同学。1. 开跑之前先把 RabbitMQ 里的几个角色认齐1.1 五个核心成员生产者、消费者、队列、交换机与绑定RabbitMQ 再复杂底层就是三样东西在干活消息、队列、交换机。生产者只负责把消息扔给交换机队列负责真正存储消息直到消费者来取消费者则从队列里拿消息处理。至于交换机你可以把它看作一个快递中转站它的任务是把收到的包裹分派到正确的队列而“分派规则”就是绑定关系。绑定关系说的直白一点就是“哪条路由键匹配的消息应该进哪个队列”的白名单。我遇到过不少初学者第一遍看代码时只盯着生产者和消费者把交换机这层给忽略了结果后面一接触 fanout / direct / topic 直接懵。我给的建议是先记住一个结论生产者从不直接把消息塞进队列所有消息都是先到交换机再由交换机根据绑定规则投递到队列。哪怕你用的是最简单的默认交换机消息也经历了这层中转。把这个链路刻进脑子里后面所有模型都会变得顺理成章。另外还要区分两个概念——连接Connection和信道Channel。Connection 是一个 TCP 长连接Channel 是建在连接内部的一条逻辑通道。生产环境里不会每条消息都新建连接而是用一个连接池维护多个 Connection每个线程从连接里取自己的 Channel 来用。Demo 里可以随手创建但你要清楚这种写法只是为了教学清晰。1.2 Windows 安装、改端口与启动失败排错很多入门教程默认你在 Linux 环境可实际上在 Windows 上玩 RabbitMQ 的人一点也不少。Windows 上安装说穿了就两步先装 Erlang再装 RabbitMQ。但这中间埋着一个最经典的坑Erlang 版本必须和 RabbitMQ 版本匹配。RabbitMQ 官方对每个版本都有对应的 Erlang 版本区间表装了一个太新的 Erlang 或太老的 Erlang服务起来后各种幺蛾子就会找上门。安装地址就直接去官网下载安装包安装时尽量用默认路径避免后面环境变量对不上。装好之后服务一般会自动注册到 Windows 服务管理器里。如果没起来打开管理员命令行进入 RabbitMQ 的 sbin 目录依次执行rabbitmq-service.bat install rabbitmq-service.bat start rabbitmq-plugins.bat enable rabbitmq_management执行完最后一条命令浏览器访问http://localhost:15672用默认账号guest/guest就能进入管理后台。注意guest账号默认只允许从本机登录如果你在局域网里想从另一台机器访问管理台需要另外创建账号并配置权限。启动失败是 Windows 上问得最多的问题我整理过一套排查顺序先看日志日志在%APPDATA%\RabbitMQ\log\目录下文件名一般是rabbit主机名.log再检查 Erlang 版本是否匹配然后确认 Windows 主机名里没有中文、空格或特殊字符RabbitMQ 节点名是带主机名的主机名不规范很容易启动中断最后看 5672 端口有没有被别的程序占用。改端口也是常见的需求。在%APPDATA%\RabbitMQ\rabbitmq.conf里写入下面的配置listeners.tcp.default 5673 management.tcp.port 15673改完重启服务就会生效。如果你的本机 5672 被其他服务占了这个配置能帮上大忙不需要重装任何东西。1.3 Linux 部署要留意的版本对应问题Linux 下部署一般是下载官方提供的.deb/.rpm包或者配置官方 APT/Yum 源后安装。安装之前先把 Erlang 装好并对照官方兼容表确认版本。很多发行版的系统源里自带的 Erlang 版本偏旧直接装上去跑最新版 RabbitMQ 十有八九会报错。装完之后记得把服务设置成开机自启sudo systemctl enable rabbitmq-server sudo systemctl start rabbitmq-server sudo rabbitmqctl statusLinux 下的日志默认在/var/log/rabbitmq/配置文件在/etc/rabbitmq/rabbitmq.conf。排错思路和 Windows 大同小异先看日志再看端口思路永远比命令重要。这套基础跑通之后我们再一个模型一个模型地过。2. 模型一简单队列——用带注释的代码跑通第一条消息2.1 简单队列到底是个什么样的结构第一个模型在官方文档里叫 Hello World是最基础的一对一模型。结构就是生产者发一条消息到队列消费者从队列里取出来。这个模型里表面上没有交换机其实底层还是走了默认交换机——路由键恰好等于队列名所以看起来就跟直接发到队列一样。这个模型的价值在于演示消息链路的最小闭环连接、声明队列、发布、消费。它的局限也很明显只有一个消费者在消费且消息没有路由能力所以生产环境基本不会拿它直接扛业务但它是最好的入门切口。2.2 生产者代码注释版import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; public class Send { // 队列名是全局的多个进程使用同一个名字才能定位到同一个队列 private final static String QUEUE_NAME hello; public static void main(String[] args) throws Exception { // 第一步创建连接工厂本质上是把 host、port、vhost、账号信息打包 ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); factory.setPort(5672); factory.setUsername(guest); factory.setPassword(guest); // 第二步建立 TCP 连接并创建 Channel // Connection 可理解为一个 TCP 长连接Channel 是里面的逻辑通道 try (Connection connection factory.newConnection(); Channel channel connection.createChannel()) { // 第三步声明队列 // 参数依次是队列名、durable是否持久化、exclusive是否独享、autoDelete无人使用时是否自动删除、扩展参数 // 如果队列不存在RabbitMQ 会创建如果已存在这一步是幂等确认 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 第四步发布消息 // exchange 传空字符串表示使用默认交换机 // routingKey 传队列名消息就会被投递到这个队列 String message Hello RabbitMQ!; channel.basicPublish(, QUEUE_NAME, null, message.getBytes(UTF-8)); System.out.println( [x] Sent message ); } // 使用 try-with-resources连接和信道会自动关闭 } }注意一个细节queueDeclare的第二个参数durable这里先给false。它代表 RabbitMQ 重启后队列是否还在对持久化的完整讨论我们后面专门展开。刚入门时不要在这个词上钻牛角尖先把消息发出去收回来。2.3 消费者代码注释版import com.rabbitmq.client.*; public class Recv { private final static String QUEUE_NAME hello; public static void main(String[] args) throws Exception { // 前两步与生产者完全一致拿到连接和信道 ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); Connection connection factory.newConnection(); Channel channel connection.createChannel(); // 消费者也要声明队列这一步不能省 // 如果生产者先启动队列已存在这里只是幂等确认 // 如果消费者先启动这一步等于把队列创建出来 channel.queueDeclare(QUEUE_NAME, false, false, false, null); System.out.println( [*] Waiting for messages...); // 注册消息回调RabbitMQ 把消息推到客户端后自动调用 DeliverCallback deliverCallback (consumerTag, delivery) - { String message new String(delivery.getBody(), UTF-8); System.out.println( [x] Received message ); }; // 参数依次是队列名、autoAck是否自动确认、消息回调、消费取消回调 // autoAcktrue 表示 RabbitMQ 把消息推出来就算消费成功 channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag - { }); } }这段代码看起来就停在basicConsume程序为什么不退出因为客户端内部启动了消费线程主线程会阻塞等待回调触发。这也是事件驱动模型最典型的样子不是消费者自己去队列里拉消息而是 RabbitMQ 把消息主动推给消费者。2.4 动手前必须搞明白的三个问题第一为什么连接不放在循环里因为创建 Connection 要建立 TCP 连接成本远高于从已有连接里开一个 Channel。在 Demo 里无所谓但真实项目里频繁建连会被服务端限制这是新手最容易踩的第一个性能坑。第二哪些场景适合用这个模型适合“只有一个生产者、一个消费者、消息只需要排队处理”的极简场景。比如补偿任务发送一条通知。一旦你有多个消费者抢同一队列的任务或者需要按级别过滤消息这个模型就撑不住了也就引出下面的工作队列。第三autoAcktrue在真实项目里为什么危险因为消息推给消费者后RabbitMQ 立刻把它从队列里删掉。如果消费者的处理逻辑还没执行就崩溃了消息就再无重试机会。后面的工作队列和路由模型我们会把autoAck改成false用代码手动确认。3. 模型二工作队列——为什么默认轮询会把任务全压给慢消费者3.1 从“一个人干活”到“一组人干活”工作队列的拓扑和简单队列几乎一样仍然使用默认交换机区别在于队列后面挂了一组消费者大家共同消费同一个队列里的任务。这个模型专门解决任务积压问题一条条消息排进队列多个消费者并行处理整体吞吐自然就上去了。典型场景是耗时任务分发比如批量发邮件、生成报表、处理上传的图片。你肯定不想让一个消费者串行处理上千张图片而是希望开十个 Worker 一起干。3.2 默认轮询机制就是一个隐藏陷阱在没有额外配置的情况下RabbitMQ 会把队列里的消息按顺序轮流发给每个消费者第一条给消费者 A第二条给消费者 B第三条又给 A。这种分发方式叫轮询看起来公平实际操作中却有一个大坑它完全不考虑每个消费者当前是否忙得过来。举个例子两个消费者共用一个队列。A 处理一条消息只需要一秒钟B 处理一条消息需要十秒。轮询模式下第 1 条给 A、第 2 条给 B、第 3 条又给 AB 永远慢吞吞地堆积A 则不断被塞新任务最后任务全堵在慢消费者手里快的反而没事做。RabbitMQ 官方把这种现象叫“循环分发”解决办法是让消费者向 RabbitMQ 声明我一次只拿一条等我处理完确认了再给我下一条。这个声明就是basicQos(1)。3.3 工作队列完整代码手动确认加预取值生产者的代码和简单队列差别不大重点是声明队列时第二个参数变成true消息也带上持久化标记public class NewTask { private static final String TASK_QUEUE_NAME task_queue; public static void main(String[] args) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); try (Connection connection factory.newConnection(); Channel channel connection.createChannel()) { // 队列设置为 durabletrueRabbitMQ 重启后队列不丢 channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null); String message String.join( , args); // 消息也标记为持久化 channel.basicPublish( , TASK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(UTF-8)); System.out.println( [x] Sent message ); } } }消费者的重点是basicQos(1)和手动basicAckpublic class Worker { private static final String TASK_QUEUE_NAME task_queue; public static void main(String[] args) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); Connection connection factory.newConnection(); Channel channel connection.createChannel(); channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null); System.out.println( [*] Waiting for tasks...); // 预取一条同一时间只给当前消费者推送一条消息 // 等消费者确认处理完成再推下一条实现“能者多劳” channel.basicQos(1); DeliverCallback deliverCallback (consumerTag, delivery) - { String message new String(delivery.getBody(), UTF-8); System.out.println( [x] Received task: message); try { // 模拟一个耗时任务比如生成缩略图 Thread.sleep(3000); // 只有真正处理完成才手动确认 channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); System.out.println( [x] Done); } catch (InterruptedException e) { Thread.currentThread().interrupt(); // 处理失败让消息回到队列尾部下次重新消费 channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true); } }; // autoAckfalseRabbitMQ 不会在推送后立即删消息必须等我们确认 channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, consumerTag - { }); } }这里有几个容易搞错的点。basicQos(1)必须放在basicConsume之前否则预取值不会生效basicAck的第二个参数multiple一般设为false表示只确认当前这条而不是批量确认。确认语义一旦理解不到位后面就会出现消息重复消费或丢失的诡异问题。3.4 ack 选错可能导致的三种后果为了把 ack 讲透我列一个实际场景对比一个消费者拿到一条消息程序在处理到一半时崩溃。根据确认方式不同结果迥异确认方式程序崩溃后会发生什么表现autoAcktrue消息在推送时已被视为消费完成消息丢失最危险autoAckfalse且从未调用 ack消息仍在队列里会重新投递给下一个消费者可能重复处理但不会丢失autoAckfalse处理完再 ack消息被正常确认删除正确做法兼顾效率和可靠性所以在工作队列模式下我强烈建议一直使用手动确认。重复消费的问题靠接口幂等性去解决这属于消费端设计问题而不是靠随意确认消息就能绕开的。4. 模型三发布/订阅——交换机第一次正式登场4.1 为什么简单队列满足不了“一条消息发给所有人”先看一个场景用户下单成功系统需要同时给用户发短信、推送 App 通知、写日志、更新积分。如果把所有逻辑都揉进下单接口代码耦合不说任何一个下游服务挂了还会拖垮主流程。正确思路是把“下单成功”这个消息广播出去谁感兴趣谁来订阅。但简单队列做不到这种广播。就算你把同一条消息复制五份分别发到五个队列你仍然要提前知道所有队列的名字而且生产者代码就得每加一个订阅方改一次完全不现实。广播需求的本质是一条消息进入交换机后交换机把它复制到所有订阅的队列。这就是 fanout 交换机的核心价值。4.2 fanout 交换机临时队列绑定关系与临时队列的作用fanout 翻译过来就是“扇出”它工作起来像一台复印机不管路由键是什么交换机收到的每条消息都会复制一份发给自己绑定的每一个队列。在发布/订阅模型里路由键完全没有意义所以代码里一律传空字符串。这个模型里还有一个新概念叫临时队列。消费者调用channel.queueDeclare()且不传任何参数时RabbitMQ 会随机生成一个队列名并自动把队列设置为 exclusive当前连接独享和 autoDelete连接关闭后队列自动删除。这样每个消费者都拥有一个只属于自己的临时队列互不干扰。临时队列非常适合广播场景短信服务和通知服务各自声明一个临时队列同时绑定到同一个 fanout 交换机“下单成功”消息就会被复制成两份分别进入两个队列。如果某个服务下线了它的临时队列自动消失不再接收消息。4.3 发布订阅模型完整代码注释生产者只需要声明 fanout 交换机然后向交换机发消息public class EmitLog { private static final String EXCHANGE_NAME logs; public static void main(String[] args) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); try (Connection connection factory.newConnection(); Channel channel connection.createChannel()) { // 声明一个 fanout 类型的交换机 // 参数依次是交换机名、类型、durable、autoDelete、扩展参数 channel.exchangeDeclare(EXCHANGE_NAME, fanout); String message info: Today is a nice day; // 注意发布目标是交换机不是队列 // fanout 交换机不区分路由键这里统一传空字符串 channel.basicPublish(EXCHANGE_NAME, , null, message.getBytes(UTF-8)); System.out.println( [x] Sent message ); } } }消费者的代码核心是“声明临时队列 绑定交换机”public class ReceiveLogs { private static final String EXCHANGE_NAME logs; public static void main(String[] args) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); Connection connection factory.newConnection(); Channel channel connection.createChannel(); // 消费者同样需要声明交换机幂等操作 channel.exchangeDeclare(EXCHANGE_NAME, fanout); // 声明临时队列不传任何参数 // RabbitMQ 会自动生成队列名并设置 exclusive 和 autoDelete String queueName channel.queueDeclare().getQueue(); // 把临时队列绑定到交换机上fanout 模式下路由键忽略 channel.queueBind(queueName, EXCHANGE_NAME, ); DeliverCallback deliverCallback (consumerTag, delivery) - { String message new String(delivery.getBody(), UTF-8); System.out.println( [x] Received: message); }; channel.basicConsume(queueName, true, deliverCallback, consumerTag - { }); } }这段代码放到生产环境最直接的感受是启动两个ReceiveLogs进程然后运行生产者两个进程都会收到同一条消息。这就是“扇出”的含义。4.4 为什么消费者必须先启动才能收到消息发布/订阅模型里有一个著名的困惑为什么我先启动生产者发消息再启动消费者消费者什么也收不到原因在于消息的生命周期。fanout 交换机只是转发器它本身不存储消息。当生产者的消息到达交换机时如果没有任何队列绑定到这个交换机消息就没有转发的去处直接被丢弃。临时队列是在消费者启动的那一刻才创建并绑定的所以消费者不在线消息自然就没了。如果你需要“消费者不在线消息也不丢”的效果就不能完全照搬这个模型而是要把队列声明为持久化 queue并用后面讲到的路由模型把消息固定投到某个队列里。理解这一点之后你就能想明白发布/订阅适合实时通知类场景但不适合必须可靠送达的异步任务。5. 模型四路由——用路由键把广播升级成定向5.1 Direct 交换机与 routing key 的匹配逻辑发布/订阅模型有一个问题它分不清消息类型全量发送。如果日志系统只想接收 error 级别的消息它还是要被迫接收所有 info、warning消费端还得自己过滤一遍。这显然浪费于是路由模型登场。路由模型使用 direct 类型的交换机。direct 的含义是精确匹配生产者发送消息时带一个路由键消费者绑定队列时带一个绑定键只有当两者完全相同时消息才会进入这个队列。这里我要把两个词分开说清楚生产者那边叫 routing key消费者这边叫 binding key。它们虽然经常是同一个字符串但概念角色不同。你可以把 binding key 理解为“队列订阅的门牌号”routing key 是“消息自己要送去的门牌号”门牌号对上了消息才进门。5.2 路由模型完整代码注释生产者按日志级别发送消息public class EmitLogDirect { private static final String EXCHANGE_NAME direct_logs; public static void main(String[] args) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); try (Connection connection factory.newConnection(); Channel channel connection.createChannel()) { // 声明 direct 交换机 channel.exchangeDeclare(EXCHANGE_NAME, direct); // 从命令行取路由键默认 info你可以运行 error、warning、info 三种级别 String routingKey args.length 0 ? args[0] : info; String message A log message with level routingKey; // 发布时带上路由键direct 交换机根据它决定投递 channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes(UTF-8)); System.out.println( [x] Sent routingKey : message ); } } }消费者按自己关心的级别绑定队列public class ReceiveLogsDirect { private static final String EXCHANGE_NAME direct_logs; public static void main(String[] args) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); Connection connection factory.newConnection(); Channel channel connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, direct); String queueName channel.queueDeclare().getQueue(); // 一个队列可以绑定多个路由键 // 下面表示这个消费者同时接收 error 和 warning 级别的日志 // 只有完全匹配的路由键消息才会进入队列 String[] severities {error, warning}; for (String severity : severities) { channel.queueBind(queueName, EXCHANGE_NAME, severity); } DeliverCallback deliverCallback (consumerTag, delivery) - { String message new String(delivery.getBody(), UTF-8); System.out.println( [x] Received delivery.getEnvelope().getRoutingKey() : message); }; channel.basicConsume(queueName, true, deliverCallback, consumerTag - { }); } }这里你会注意到绑定发生在队列和交换机之间数据模型从“队列”扩展到了“交换机 绑定关系”。这也是 RabbitMQ 和普通 Redis List 队列最大的差异——消息不是直接进队列而是先经过一层可配置的路由。5.3 路由键和绑定键最容易踩的三个坑第一个坑大小写敏感。error和Error是两个完全不同的键生产环境里经常因为大小写不统一导致消息静默丢失排查半天才发现是手滑。第二个坑消息发布到交换机时如果没有任何队列绑定了对应的路由键这条消息会被交换机直接丢弃不会报错也没有日志。很多人第一次排查这种问题时会觉得莫名其妙记住这一点能省很多时间。第三个坑一个交换机只能有一种类型你不能创建direct_logs时它叫 direct后来又想给它加一个 fanout 绑定。交换机类型在声明时就固定了想换类型只能重新建一个交换机。规划名字时最好把类型带进去比如exchange.direct.logs这样看名字就知道用途。6. 模型五主题——用两个通配符换来最灵活的路由6.1 星号和井号的匹配规则主题模型使用 topic 交换机它和 direct 的差别在于direct 只能精确匹配topic 支持 fuzzy 匹配。消息的路由键是多个单词用点号连接比如lazy.orange.rabbit消费者的绑定键里可以使用两个通配符*匹配一个单词#匹配零个或多个单词所以*.orange.*能匹配quick.orange.rabbit因为它表示“第一个单词任意、第二个单词必须是 orange、第三个单词任意”。而lazy.#表示“第一个单词必须是 lazy后面随便有多少个单词都行”。理解这个规则的窍门是把路由键想成用点分隔的若干段“*”占一个段位“#”占剩余所有段位。6.2 主题模型完整代码注释生产者只需要声明 topic 交换机并发送带路由键的消息public class EmitLogTopic { private static final String EXCHANGE_NAME topic_logs; public static void main(String[] args) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); try (Connection connection factory.newConnection(); Channel channel connection.createChannel()) { // 声明 topic 交换机 channel.exchangeDeclare(EXCHANGE_NAME, topic); // 路由键可以是 lazy.orange.rabbit、quick.orange.fox 等 String routingKey lazy.orange.rabbit; String message A lazy orange rabbit lives here; channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes(UTF-8)); System.out.println( [x] Sent routingKey : message ); } } }消费者的绑定键支持通配符可以精确把握订阅范围public class ReceiveLogsTopic { private static final String EXCHANGE_NAME topic_logs; public static void main(String[] args) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); Connection connection factory.newConnection(); Channel channel connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, topic); String queueName channel.queueDeclare().getQueue(); // 一条队列可以绑定多个通配符表达式 // 下面表示我对所有“中间段是 orange”的消息感兴趣 // 同时对所有“lazy 开头”的消息感兴趣 channel.queueBind(queueName, EXCHANGE_NAME, *.orange.*); channel.queueBind(queueName, EXCHANGE_NAME, lazy.#); DeliverCallback deliverCallback (consumerTag, delivery) - { String message new String(delivery.getBody(), UTF-8); System.out.println( [x] Received delivery.getEnvelope().getRoutingKey() : message); }; channel.basicConsume(queueName, true, deliverCallback, consumerTag - { }); } }6.3 路由匹配对照表与易错点为了让大家牢牢掌握匹配规则我把官方文档的经典案例扩展成一张对照表。假设有三个绑定键分别是*.orange.*、*.*.rabbit、lazy.#不同的路由键命中情况如下路由键发送*.orange.**.*.rabbitlazy.#quick.orange.rabbit命中命中不命中lazy.orange.rabbit命中命中命中quick.orange.fox命中不命中不命中lazy.brown.fox不命中不命中命中quick.brown.fox不命中命中不命中orange不命中不命中不命中lazy不命中不命中命中注意最后两行。orange只有一个单词*.orange.*需要三个单词所以不匹配。lazy同样只有一个单词但lazy.#中的#可以代表零个单词所以匹配。这些边界是面试题里的高频考点实战场上也很容易出现“我明明绑定了*.orange.*发了个orange怎么没反应”的困惑。主题模型最容易踩的坑有三类一是绑定键里混入了空格比如*.orange.*尾部多个空格整条规则就坏了二是路由键中不该有空格路由键只能用点号分隔单词不是用空格三是#放的位置不同语义不同#.lazy只匹配以lazy结尾的键lazy.#只匹配以lazy开头的键不要混用。7. 五大模型怎么选以及从 Demo 到真实项目最容易忽略的三件事7.1 模型选型速查表五种模型并不是并列的五种工具而是层层递进的关系。我把它们整理成一张表方便你按照业务需求对号入座模型交换机类型路由方式典型场景简单队列默认交换机路由键等于队列名入门演示、单点消费工作队列默认交换机路由键等于队列名任务分发、多消费者并联处理发布/订阅fanout广播忽略路由键事件广播、实时通知路由direct精确匹配路由键按级别过滤日志主题topic通配符匹配路由键多维订阅、复杂事件分发实际选型时我一般这样判断消息只需要一个消费者处理那就用工作队列需要同时推给多方用发布/订阅需要按不同类别分发给不同消费者先看类别是有限且固定的那就用路由如果类别有层级、订阅条件复杂直接用主题。大部分系统最常用的其实是工作队列和主题两种前者负责削峰后者负责灵活路由。7.2 RabbitMQ 和 Kafka、RocketMQ 的边界在哪里讲完五大模型很多读者会自然想到一个问题选型对比时RabbitMQ 到底适合什么不适合什么我的观点是RabbitMQ 的核心优势是它拥有最灵活的消息模型和稳定的 AMQP 协议生态适合路由规则复杂、消费方式多样、对消息可靠性和即时性要求高的系统。Kafka 则强在海量日志、高吞吐、顺序消费和流处理它把消息模型简化成了分区加偏移量这条路径更适合做数据管道。RocketMQ 在电商大促场景下很成熟事务消息和对消息结果的精细控制也是它的长板。如果你的主要需求就是“100 个消费者订阅 100 种事件不同事件走不同逻辑”RabbitMQ 的主题交换机会让你写起来非常顺手。如果你追求的是每秒几十万条日志的吞吐量那它就不是最优解。聊选型不是键盘侠对喷而是先把自己的场景列清楚再看哪个中间件离场景最近。7.3 生产环境补强连接复用、持久化与死信队列先把一个残酷的事实说在前面官方五大模型 Demo 全部照抄跑通了也还不能上生产。至少要知道三块补强工程。连接复用与异常重连。Demo 里每次 main 方法都创建 Connection 关闭 Connection生产环境必须用连接池并且要监控断线重连。RabbitMQ 客户端提供自动恢复机制默认情况下如果网络异常闪断连接会自动恢复重建。但如果你自己的业务代码没处理好信道复用恢复后可能出现信道已关闭的异常所以尽量保持一个 Channel 只在一个线程里使用别跨线程共享。持久化不是一句话的事。队列持久化、交换机持久化、消息持久化这三层缺一不可。队列和交换机声明时把 durable 设为 true消息发布时使用持久化属性broker 重启后才有可能恢复到消息。但要注意持久化并不能保证消息绝对不丢它只能应付“RabbitMQ 重启”这类正常故障如果机器直接断电最后几毫秒仍在内存里的消息依然可能丢失。死信队列是生产环境绕不开的机制。所谓死信就是消息因为被拒绝、过期、或队列长度超限而被丢弃之前被转发到另一个交换机。声明队列时通过参数指定MapString, Object args new HashMap(); args.put(x-dead-letter-exchange, dlx.exchange); args.put(x-dead-letter-routing-key, dlx.routing.key); channel.queueDeclare(main.queue, true, false, false, args);这样主队列里无法被正常消费的消息会进入死信交换机再由死信交换机投到对应的死信队列方便你集中排查那些“怎么都消费不掉”的消息。配合延迟队列插件和消息 TTL还能实现定时任务、重试补偿等能力。我在实际项目里见过不少事故起因往往不是某个高级特性没掌握而是最基础的模型理解偏了。发布/订阅模型被拿去当工作队列用临时队列被用在必须保证消息不丢的场景topic 绑定键写错导致线上静默丢消息这些都属于“模型选错加概念混淆”。所以如果你还在入门阶段别急着直接套 Spring Boot 封装先把五种模型用最小代码各跑一遍跑通之后再回头看框架你会顺手得多。 RabbitMQ五大消息模型详解：交换机、路由与实战代码精讲
做后端开发这几年RabbitMQ 算是我最常用也最常被问到的一个消息中间件。网上资料不少但很多朋友卡在第一步五大消息模型到底是什么为什么有的模型需要交换机有的模型只靠队列就能跑通这篇文章就是把基础彻底补上——按官方 Tutorial 的脉络把简单队列、工作队列、发布/订阅、路由、主题这五种模型拆开讲一遍代码全部带详细注释顺手把 Windows 和 Linux 下安装、启动失败、改端口这些实战坑也一起填了。适合刚接触 RabbitMQ 的读者也适合已经用过一段时间、但一直没把模型关系真正理清的同学。1. 开跑之前先把 RabbitMQ 里的几个角色认齐1.1 五个核心成员生产者、消费者、队列、交换机与绑定RabbitMQ 再复杂底层就是三样东西在干活消息、队列、交换机。生产者只负责把消息扔给交换机队列负责真正存储消息直到消费者来取消费者则从队列里拿消息处理。至于交换机你可以把它看作一个快递中转站它的任务是把收到的包裹分派到正确的队列而“分派规则”就是绑定关系。绑定关系说的直白一点就是“哪条路由键匹配的消息应该进哪个队列”的白名单。我遇到过不少初学者第一遍看代码时只盯着生产者和消费者把交换机这层给忽略了结果后面一接触 fanout / direct / topic 直接懵。我给的建议是先记住一个结论生产者从不直接把消息塞进队列所有消息都是先到交换机再由交换机根据绑定规则投递到队列。哪怕你用的是最简单的默认交换机消息也经历了这层中转。把这个链路刻进脑子里后面所有模型都会变得顺理成章。另外还要区分两个概念——连接Connection和信道Channel。Connection 是一个 TCP 长连接Channel 是建在连接内部的一条逻辑通道。生产环境里不会每条消息都新建连接而是用一个连接池维护多个 Connection每个线程从连接里取自己的 Channel 来用。Demo 里可以随手创建但你要清楚这种写法只是为了教学清晰。1.2 Windows 安装、改端口与启动失败排错很多入门教程默认你在 Linux 环境可实际上在 Windows 上玩 RabbitMQ 的人一点也不少。Windows 上安装说穿了就两步先装 Erlang再装 RabbitMQ。但这中间埋着一个最经典的坑Erlang 版本必须和 RabbitMQ 版本匹配。RabbitMQ 官方对每个版本都有对应的 Erlang 版本区间表装了一个太新的 Erlang 或太老的 Erlang服务起来后各种幺蛾子就会找上门。安装地址就直接去官网下载安装包安装时尽量用默认路径避免后面环境变量对不上。装好之后服务一般会自动注册到 Windows 服务管理器里。如果没起来打开管理员命令行进入 RabbitMQ 的 sbin 目录依次执行rabbitmq-service.bat install rabbitmq-service.bat start rabbitmq-plugins.bat enable rabbitmq_management执行完最后一条命令浏览器访问http://localhost:15672用默认账号guest/guest就能进入管理后台。注意guest账号默认只允许从本机登录如果你在局域网里想从另一台机器访问管理台需要另外创建账号并配置权限。启动失败是 Windows 上问得最多的问题我整理过一套排查顺序先看日志日志在%APPDATA%\RabbitMQ\log\目录下文件名一般是rabbit主机名.log再检查 Erlang 版本是否匹配然后确认 Windows 主机名里没有中文、空格或特殊字符RabbitMQ 节点名是带主机名的主机名不规范很容易启动中断最后看 5672 端口有没有被别的程序占用。改端口也是常见的需求。在%APPDATA%\RabbitMQ\rabbitmq.conf里写入下面的配置listeners.tcp.default 5673 management.tcp.port 15673改完重启服务就会生效。如果你的本机 5672 被其他服务占了这个配置能帮上大忙不需要重装任何东西。1.3 Linux 部署要留意的版本对应问题Linux 下部署一般是下载官方提供的.deb/.rpm包或者配置官方 APT/Yum 源后安装。安装之前先把 Erlang 装好并对照官方兼容表确认版本。很多发行版的系统源里自带的 Erlang 版本偏旧直接装上去跑最新版 RabbitMQ 十有八九会报错。装完之后记得把服务设置成开机自启sudo systemctl enable rabbitmq-server sudo systemctl start rabbitmq-server sudo rabbitmqctl statusLinux 下的日志默认在/var/log/rabbitmq/配置文件在/etc/rabbitmq/rabbitmq.conf。排错思路和 Windows 大同小异先看日志再看端口思路永远比命令重要。这套基础跑通之后我们再一个模型一个模型地过。2. 模型一简单队列——用带注释的代码跑通第一条消息2.1 简单队列到底是个什么样的结构第一个模型在官方文档里叫 Hello World是最基础的一对一模型。结构就是生产者发一条消息到队列消费者从队列里取出来。这个模型里表面上没有交换机其实底层还是走了默认交换机——路由键恰好等于队列名所以看起来就跟直接发到队列一样。这个模型的价值在于演示消息链路的最小闭环连接、声明队列、发布、消费。它的局限也很明显只有一个消费者在消费且消息没有路由能力所以生产环境基本不会拿它直接扛业务但它是最好的入门切口。2.2 生产者代码注释版import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; public class Send { // 队列名是全局的多个进程使用同一个名字才能定位到同一个队列 private final static String QUEUE_NAME hello; public static void main(String[] args) throws Exception { // 第一步创建连接工厂本质上是把 host、port、vhost、账号信息打包 ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); factory.setPort(5672); factory.setUsername(guest); factory.setPassword(guest); // 第二步建立 TCP 连接并创建 Channel // Connection 可理解为一个 TCP 长连接Channel 是里面的逻辑通道 try (Connection connection factory.newConnection(); Channel channel connection.createChannel()) { // 第三步声明队列 // 参数依次是队列名、durable是否持久化、exclusive是否独享、autoDelete无人使用时是否自动删除、扩展参数 // 如果队列不存在RabbitMQ 会创建如果已存在这一步是幂等确认 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 第四步发布消息 // exchange 传空字符串表示使用默认交换机 // routingKey 传队列名消息就会被投递到这个队列 String message Hello RabbitMQ!; channel.basicPublish(, QUEUE_NAME, null, message.getBytes(UTF-8)); System.out.println( [x] Sent message ); } // 使用 try-with-resources连接和信道会自动关闭 } }注意一个细节queueDeclare的第二个参数durable这里先给false。它代表 RabbitMQ 重启后队列是否还在对持久化的完整讨论我们后面专门展开。刚入门时不要在这个词上钻牛角尖先把消息发出去收回来。2.3 消费者代码注释版import com.rabbitmq.client.*; public class Recv { private final static String QUEUE_NAME hello; public static void main(String[] args) throws Exception { // 前两步与生产者完全一致拿到连接和信道 ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); Connection connection factory.newConnection(); Channel channel connection.createChannel(); // 消费者也要声明队列这一步不能省 // 如果生产者先启动队列已存在这里只是幂等确认 // 如果消费者先启动这一步等于把队列创建出来 channel.queueDeclare(QUEUE_NAME, false, false, false, null); System.out.println( [*] Waiting for messages...); // 注册消息回调RabbitMQ 把消息推到客户端后自动调用 DeliverCallback deliverCallback (consumerTag, delivery) - { String message new String(delivery.getBody(), UTF-8); System.out.println( [x] Received message ); }; // 参数依次是队列名、autoAck是否自动确认、消息回调、消费取消回调 // autoAcktrue 表示 RabbitMQ 把消息推出来就算消费成功 channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag - { }); } }这段代码看起来就停在basicConsume程序为什么不退出因为客户端内部启动了消费线程主线程会阻塞等待回调触发。这也是事件驱动模型最典型的样子不是消费者自己去队列里拉消息而是 RabbitMQ 把消息主动推给消费者。2.4 动手前必须搞明白的三个问题第一为什么连接不放在循环里因为创建 Connection 要建立 TCP 连接成本远高于从已有连接里开一个 Channel。在 Demo 里无所谓但真实项目里频繁建连会被服务端限制这是新手最容易踩的第一个性能坑。第二哪些场景适合用这个模型适合“只有一个生产者、一个消费者、消息只需要排队处理”的极简场景。比如补偿任务发送一条通知。一旦你有多个消费者抢同一队列的任务或者需要按级别过滤消息这个模型就撑不住了也就引出下面的工作队列。第三autoAcktrue在真实项目里为什么危险因为消息推给消费者后RabbitMQ 立刻把它从队列里删掉。如果消费者的处理逻辑还没执行就崩溃了消息就再无重试机会。后面的工作队列和路由模型我们会把autoAck改成false用代码手动确认。3. 模型二工作队列——为什么默认轮询会把任务全压给慢消费者3.1 从“一个人干活”到“一组人干活”工作队列的拓扑和简单队列几乎一样仍然使用默认交换机区别在于队列后面挂了一组消费者大家共同消费同一个队列里的任务。这个模型专门解决任务积压问题一条条消息排进队列多个消费者并行处理整体吞吐自然就上去了。典型场景是耗时任务分发比如批量发邮件、生成报表、处理上传的图片。你肯定不想让一个消费者串行处理上千张图片而是希望开十个 Worker 一起干。3.2 默认轮询机制就是一个隐藏陷阱在没有额外配置的情况下RabbitMQ 会把队列里的消息按顺序轮流发给每个消费者第一条给消费者 A第二条给消费者 B第三条又给 A。这种分发方式叫轮询看起来公平实际操作中却有一个大坑它完全不考虑每个消费者当前是否忙得过来。举个例子两个消费者共用一个队列。A 处理一条消息只需要一秒钟B 处理一条消息需要十秒。轮询模式下第 1 条给 A、第 2 条给 B、第 3 条又给 AB 永远慢吞吞地堆积A 则不断被塞新任务最后任务全堵在慢消费者手里快的反而没事做。RabbitMQ 官方把这种现象叫“循环分发”解决办法是让消费者向 RabbitMQ 声明我一次只拿一条等我处理完确认了再给我下一条。这个声明就是basicQos(1)。3.3 工作队列完整代码手动确认加预取值生产者的代码和简单队列差别不大重点是声明队列时第二个参数变成true消息也带上持久化标记public class NewTask { private static final String TASK_QUEUE_NAME task_queue; public static void main(String[] args) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); try (Connection connection factory.newConnection(); Channel channel connection.createChannel()) { // 队列设置为 durabletrueRabbitMQ 重启后队列不丢 channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null); String message String.join( , args); // 消息也标记为持久化 channel.basicPublish( , TASK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(UTF-8)); System.out.println( [x] Sent message ); } } }消费者的重点是basicQos(1)和手动basicAckpublic class Worker { private static final String TASK_QUEUE_NAME task_queue; public static void main(String[] args) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); Connection connection factory.newConnection(); Channel channel connection.createChannel(); channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null); System.out.println( [*] Waiting for tasks...); // 预取一条同一时间只给当前消费者推送一条消息 // 等消费者确认处理完成再推下一条实现“能者多劳” channel.basicQos(1); DeliverCallback deliverCallback (consumerTag, delivery) - { String message new String(delivery.getBody(), UTF-8); System.out.println( [x] Received task: message); try { // 模拟一个耗时任务比如生成缩略图 Thread.sleep(3000); // 只有真正处理完成才手动确认 channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); System.out.println( [x] Done); } catch (InterruptedException e) { Thread.currentThread().interrupt(); // 处理失败让消息回到队列尾部下次重新消费 channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true); } }; // autoAckfalseRabbitMQ 不会在推送后立即删消息必须等我们确认 channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, consumerTag - { }); } }这里有几个容易搞错的点。basicQos(1)必须放在basicConsume之前否则预取值不会生效basicAck的第二个参数multiple一般设为false表示只确认当前这条而不是批量确认。确认语义一旦理解不到位后面就会出现消息重复消费或丢失的诡异问题。3.4 ack 选错可能导致的三种后果为了把 ack 讲透我列一个实际场景对比一个消费者拿到一条消息程序在处理到一半时崩溃。根据确认方式不同结果迥异确认方式程序崩溃后会发生什么表现autoAcktrue消息在推送时已被视为消费完成消息丢失最危险autoAckfalse且从未调用 ack消息仍在队列里会重新投递给下一个消费者可能重复处理但不会丢失autoAckfalse处理完再 ack消息被正常确认删除正确做法兼顾效率和可靠性所以在工作队列模式下我强烈建议一直使用手动确认。重复消费的问题靠接口幂等性去解决这属于消费端设计问题而不是靠随意确认消息就能绕开的。4. 模型三发布/订阅——交换机第一次正式登场4.1 为什么简单队列满足不了“一条消息发给所有人”先看一个场景用户下单成功系统需要同时给用户发短信、推送 App 通知、写日志、更新积分。如果把所有逻辑都揉进下单接口代码耦合不说任何一个下游服务挂了还会拖垮主流程。正确思路是把“下单成功”这个消息广播出去谁感兴趣谁来订阅。但简单队列做不到这种广播。就算你把同一条消息复制五份分别发到五个队列你仍然要提前知道所有队列的名字而且生产者代码就得每加一个订阅方改一次完全不现实。广播需求的本质是一条消息进入交换机后交换机把它复制到所有订阅的队列。这就是 fanout 交换机的核心价值。4.2 fanout 交换机临时队列绑定关系与临时队列的作用fanout 翻译过来就是“扇出”它工作起来像一台复印机不管路由键是什么交换机收到的每条消息都会复制一份发给自己绑定的每一个队列。在发布/订阅模型里路由键完全没有意义所以代码里一律传空字符串。这个模型里还有一个新概念叫临时队列。消费者调用channel.queueDeclare()且不传任何参数时RabbitMQ 会随机生成一个队列名并自动把队列设置为 exclusive当前连接独享和 autoDelete连接关闭后队列自动删除。这样每个消费者都拥有一个只属于自己的临时队列互不干扰。临时队列非常适合广播场景短信服务和通知服务各自声明一个临时队列同时绑定到同一个 fanout 交换机“下单成功”消息就会被复制成两份分别进入两个队列。如果某个服务下线了它的临时队列自动消失不再接收消息。4.3 发布订阅模型完整代码注释生产者只需要声明 fanout 交换机然后向交换机发消息public class EmitLog { private static final String EXCHANGE_NAME logs; public static void main(String[] args) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); try (Connection connection factory.newConnection(); Channel channel connection.createChannel()) { // 声明一个 fanout 类型的交换机 // 参数依次是交换机名、类型、durable、autoDelete、扩展参数 channel.exchangeDeclare(EXCHANGE_NAME, fanout); String message info: Today is a nice day; // 注意发布目标是交换机不是队列 // fanout 交换机不区分路由键这里统一传空字符串 channel.basicPublish(EXCHANGE_NAME, , null, message.getBytes(UTF-8)); System.out.println( [x] Sent message ); } } }消费者的代码核心是“声明临时队列 绑定交换机”public class ReceiveLogs { private static final String EXCHANGE_NAME logs; public static void main(String[] args) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); Connection connection factory.newConnection(); Channel channel connection.createChannel(); // 消费者同样需要声明交换机幂等操作 channel.exchangeDeclare(EXCHANGE_NAME, fanout); // 声明临时队列不传任何参数 // RabbitMQ 会自动生成队列名并设置 exclusive 和 autoDelete String queueName channel.queueDeclare().getQueue(); // 把临时队列绑定到交换机上fanout 模式下路由键忽略 channel.queueBind(queueName, EXCHANGE_NAME, ); DeliverCallback deliverCallback (consumerTag, delivery) - { String message new String(delivery.getBody(), UTF-8); System.out.println( [x] Received: message); }; channel.basicConsume(queueName, true, deliverCallback, consumerTag - { }); } }这段代码放到生产环境最直接的感受是启动两个ReceiveLogs进程然后运行生产者两个进程都会收到同一条消息。这就是“扇出”的含义。4.4 为什么消费者必须先启动才能收到消息发布/订阅模型里有一个著名的困惑为什么我先启动生产者发消息再启动消费者消费者什么也收不到原因在于消息的生命周期。fanout 交换机只是转发器它本身不存储消息。当生产者的消息到达交换机时如果没有任何队列绑定到这个交换机消息就没有转发的去处直接被丢弃。临时队列是在消费者启动的那一刻才创建并绑定的所以消费者不在线消息自然就没了。如果你需要“消费者不在线消息也不丢”的效果就不能完全照搬这个模型而是要把队列声明为持久化 queue并用后面讲到的路由模型把消息固定投到某个队列里。理解这一点之后你就能想明白发布/订阅适合实时通知类场景但不适合必须可靠送达的异步任务。5. 模型四路由——用路由键把广播升级成定向5.1 Direct 交换机与 routing key 的匹配逻辑发布/订阅模型有一个问题它分不清消息类型全量发送。如果日志系统只想接收 error 级别的消息它还是要被迫接收所有 info、warning消费端还得自己过滤一遍。这显然浪费于是路由模型登场。路由模型使用 direct 类型的交换机。direct 的含义是精确匹配生产者发送消息时带一个路由键消费者绑定队列时带一个绑定键只有当两者完全相同时消息才会进入这个队列。这里我要把两个词分开说清楚生产者那边叫 routing key消费者这边叫 binding key。它们虽然经常是同一个字符串但概念角色不同。你可以把 binding key 理解为“队列订阅的门牌号”routing key 是“消息自己要送去的门牌号”门牌号对上了消息才进门。5.2 路由模型完整代码注释生产者按日志级别发送消息public class EmitLogDirect { private static final String EXCHANGE_NAME direct_logs; public static void main(String[] args) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); try (Connection connection factory.newConnection(); Channel channel connection.createChannel()) { // 声明 direct 交换机 channel.exchangeDeclare(EXCHANGE_NAME, direct); // 从命令行取路由键默认 info你可以运行 error、warning、info 三种级别 String routingKey args.length 0 ? args[0] : info; String message A log message with level routingKey; // 发布时带上路由键direct 交换机根据它决定投递 channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes(UTF-8)); System.out.println( [x] Sent routingKey : message ); } } }消费者按自己关心的级别绑定队列public class ReceiveLogsDirect { private static final String EXCHANGE_NAME direct_logs; public static void main(String[] args) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); Connection connection factory.newConnection(); Channel channel connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, direct); String queueName channel.queueDeclare().getQueue(); // 一个队列可以绑定多个路由键 // 下面表示这个消费者同时接收 error 和 warning 级别的日志 // 只有完全匹配的路由键消息才会进入队列 String[] severities {error, warning}; for (String severity : severities) { channel.queueBind(queueName, EXCHANGE_NAME, severity); } DeliverCallback deliverCallback (consumerTag, delivery) - { String message new String(delivery.getBody(), UTF-8); System.out.println( [x] Received delivery.getEnvelope().getRoutingKey() : message); }; channel.basicConsume(queueName, true, deliverCallback, consumerTag - { }); } }这里你会注意到绑定发生在队列和交换机之间数据模型从“队列”扩展到了“交换机 绑定关系”。这也是 RabbitMQ 和普通 Redis List 队列最大的差异——消息不是直接进队列而是先经过一层可配置的路由。5.3 路由键和绑定键最容易踩的三个坑第一个坑大小写敏感。error和Error是两个完全不同的键生产环境里经常因为大小写不统一导致消息静默丢失排查半天才发现是手滑。第二个坑消息发布到交换机时如果没有任何队列绑定了对应的路由键这条消息会被交换机直接丢弃不会报错也没有日志。很多人第一次排查这种问题时会觉得莫名其妙记住这一点能省很多时间。第三个坑一个交换机只能有一种类型你不能创建direct_logs时它叫 direct后来又想给它加一个 fanout 绑定。交换机类型在声明时就固定了想换类型只能重新建一个交换机。规划名字时最好把类型带进去比如exchange.direct.logs这样看名字就知道用途。6. 模型五主题——用两个通配符换来最灵活的路由6.1 星号和井号的匹配规则主题模型使用 topic 交换机它和 direct 的差别在于direct 只能精确匹配topic 支持 fuzzy 匹配。消息的路由键是多个单词用点号连接比如lazy.orange.rabbit消费者的绑定键里可以使用两个通配符*匹配一个单词#匹配零个或多个单词所以*.orange.*能匹配quick.orange.rabbit因为它表示“第一个单词任意、第二个单词必须是 orange、第三个单词任意”。而lazy.#表示“第一个单词必须是 lazy后面随便有多少个单词都行”。理解这个规则的窍门是把路由键想成用点分隔的若干段“*”占一个段位“#”占剩余所有段位。6.2 主题模型完整代码注释生产者只需要声明 topic 交换机并发送带路由键的消息public class EmitLogTopic { private static final String EXCHANGE_NAME topic_logs; public static void main(String[] args) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); try (Connection connection factory.newConnection(); Channel channel connection.createChannel()) { // 声明 topic 交换机 channel.exchangeDeclare(EXCHANGE_NAME, topic); // 路由键可以是 lazy.orange.rabbit、quick.orange.fox 等 String routingKey lazy.orange.rabbit; String message A lazy orange rabbit lives here; channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes(UTF-8)); System.out.println( [x] Sent routingKey : message ); } } }消费者的绑定键支持通配符可以精确把握订阅范围public class ReceiveLogsTopic { private static final String EXCHANGE_NAME topic_logs; public static void main(String[] args) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); Connection connection factory.newConnection(); Channel channel connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, topic); String queueName channel.queueDeclare().getQueue(); // 一条队列可以绑定多个通配符表达式 // 下面表示我对所有“中间段是 orange”的消息感兴趣 // 同时对所有“lazy 开头”的消息感兴趣 channel.queueBind(queueName, EXCHANGE_NAME, *.orange.*); channel.queueBind(queueName, EXCHANGE_NAME, lazy.#); DeliverCallback deliverCallback (consumerTag, delivery) - { String message new String(delivery.getBody(), UTF-8); System.out.println( [x] Received delivery.getEnvelope().getRoutingKey() : message); }; channel.basicConsume(queueName, true, deliverCallback, consumerTag - { }); } }6.3 路由匹配对照表与易错点为了让大家牢牢掌握匹配规则我把官方文档的经典案例扩展成一张对照表。假设有三个绑定键分别是*.orange.*、*.*.rabbit、lazy.#不同的路由键命中情况如下路由键发送*.orange.**.*.rabbitlazy.#quick.orange.rabbit命中命中不命中lazy.orange.rabbit命中命中命中quick.orange.fox命中不命中不命中lazy.brown.fox不命中不命中命中quick.brown.fox不命中命中不命中orange不命中不命中不命中lazy不命中不命中命中注意最后两行。orange只有一个单词*.orange.*需要三个单词所以不匹配。lazy同样只有一个单词但lazy.#中的#可以代表零个单词所以匹配。这些边界是面试题里的高频考点实战场上也很容易出现“我明明绑定了*.orange.*发了个orange怎么没反应”的困惑。主题模型最容易踩的坑有三类一是绑定键里混入了空格比如*.orange.*尾部多个空格整条规则就坏了二是路由键中不该有空格路由键只能用点号分隔单词不是用空格三是#放的位置不同语义不同#.lazy只匹配以lazy结尾的键lazy.#只匹配以lazy开头的键不要混用。7. 五大模型怎么选以及从 Demo 到真实项目最容易忽略的三件事7.1 模型选型速查表五种模型并不是并列的五种工具而是层层递进的关系。我把它们整理成一张表方便你按照业务需求对号入座模型交换机类型路由方式典型场景简单队列默认交换机路由键等于队列名入门演示、单点消费工作队列默认交换机路由键等于队列名任务分发、多消费者并联处理发布/订阅fanout广播忽略路由键事件广播、实时通知路由direct精确匹配路由键按级别过滤日志主题topic通配符匹配路由键多维订阅、复杂事件分发实际选型时我一般这样判断消息只需要一个消费者处理那就用工作队列需要同时推给多方用发布/订阅需要按不同类别分发给不同消费者先看类别是有限且固定的那就用路由如果类别有层级、订阅条件复杂直接用主题。大部分系统最常用的其实是工作队列和主题两种前者负责削峰后者负责灵活路由。7.2 RabbitMQ 和 Kafka、RocketMQ 的边界在哪里讲完五大模型很多读者会自然想到一个问题选型对比时RabbitMQ 到底适合什么不适合什么我的观点是RabbitMQ 的核心优势是它拥有最灵活的消息模型和稳定的 AMQP 协议生态适合路由规则复杂、消费方式多样、对消息可靠性和即时性要求高的系统。Kafka 则强在海量日志、高吞吐、顺序消费和流处理它把消息模型简化成了分区加偏移量这条路径更适合做数据管道。RocketMQ 在电商大促场景下很成熟事务消息和对消息结果的精细控制也是它的长板。如果你的主要需求就是“100 个消费者订阅 100 种事件不同事件走不同逻辑”RabbitMQ 的主题交换机会让你写起来非常顺手。如果你追求的是每秒几十万条日志的吞吐量那它就不是最优解。聊选型不是键盘侠对喷而是先把自己的场景列清楚再看哪个中间件离场景最近。7.3 生产环境补强连接复用、持久化与死信队列先把一个残酷的事实说在前面官方五大模型 Demo 全部照抄跑通了也还不能上生产。至少要知道三块补强工程。连接复用与异常重连。Demo 里每次 main 方法都创建 Connection 关闭 Connection生产环境必须用连接池并且要监控断线重连。RabbitMQ 客户端提供自动恢复机制默认情况下如果网络异常闪断连接会自动恢复重建。但如果你自己的业务代码没处理好信道复用恢复后可能出现信道已关闭的异常所以尽量保持一个 Channel 只在一个线程里使用别跨线程共享。持久化不是一句话的事。队列持久化、交换机持久化、消息持久化这三层缺一不可。队列和交换机声明时把 durable 设为 true消息发布时使用持久化属性broker 重启后才有可能恢复到消息。但要注意持久化并不能保证消息绝对不丢它只能应付“RabbitMQ 重启”这类正常故障如果机器直接断电最后几毫秒仍在内存里的消息依然可能丢失。死信队列是生产环境绕不开的机制。所谓死信就是消息因为被拒绝、过期、或队列长度超限而被丢弃之前被转发到另一个交换机。声明队列时通过参数指定MapString, Object args new HashMap(); args.put(x-dead-letter-exchange, dlx.exchange); args.put(x-dead-letter-routing-key, dlx.routing.key); channel.queueDeclare(main.queue, true, false, false, args);这样主队列里无法被正常消费的消息会进入死信交换机再由死信交换机投到对应的死信队列方便你集中排查那些“怎么都消费不掉”的消息。配合延迟队列插件和消息 TTL还能实现定时任务、重试补偿等能力。我在实际项目里见过不少事故起因往往不是某个高级特性没掌握而是最基础的模型理解偏了。发布/订阅模型被拿去当工作队列用临时队列被用在必须保证消息不丢的场景topic 绑定键写错导致线上静默丢消息这些都属于“模型选错加概念混淆”。所以如果你还在入门阶段别急着直接套 Spring Boot 封装先把五种模型用最小代码各跑一遍跑通之后再回头看框架你会顺手得多。