Java 中间件之 RabbitMQ 快速入门: SpringAMQP 的 DirectExchange 路由模式- 惊觉

概述

Fanout 交换机会把消息无差别地复制给所有绑定队列，一旦你需要"这条消息只给告警服务，不给归档服务"，Fanout 就不够用了。DirectExchange 用 routingKey 与 bindingKey 的精确匹配解决这件事：发送方带上 key，交换机只把消息投给 bindingKey 完全相等的队列。

纲要

Direct 与 Fanout 的差异
- Fanout：广播，忽略 key，所有绑定队列各拿一份
- Direct：定向，routingKey 与 bindingKey 全等才投递
- 多个队列绑定相同 key 时，Direct 可退化成广播
路由原理与投递结果
- routingKey / bindingKey 的对暗号模型
- 一组消息（red / blue / yellow / green）的实际落点
- 无匹配 bindingKey 时消息被静默丢弃
动手：基于 @RabbitListener 注解声明交换机、队列、绑定
- @QueueBinding / @Queue / @Exchange 三个注解
- 与 @Bean 声明式写法的对比
两种写法对比
- @Bean + BindingBuilder（FanoutConfig 那一套）
- 注解式（本文重点，代码量少一半）
典型落地场景：日志分级
- error 进告警队列，info/warn 只进归档队列
实战坑
- key 写错导致消息被丢且无报错
- 一个队列绑多个 key 的两种写法
- Direct 是精确匹配，order.* 不会被当通配符

Direct 与 Fanout 的差异

FanoutExchange 是"订阅即收"，交换机完全不看消息内容，把消息复制 N 份发给 N 个绑定队列。DirectExchange 是"对暗号"：

每个队列与交换机绑定时都要声明一个 bindingKey，一个队列可以声明多个
生产者发送消息时必须带上 routingKey
交换机逐个比对，routingKey.equals(bindingKey) 成立才投递

维度	FanoutExchange	DirectExchange
路由依据	不看 key	routingKey 与 bindingKey 全等
一条消息的落点	所有绑定队列	匹配到的队列（0 个到 N 个）
广播能力	天然广播	所有队列绑同一个 key 时可以模拟广播
发送方负担	`routingKey` 填空串即可	必须正确填写 routingKey
灵活性	低	高，代价是多维护一份 key 约定

反过来说，Direct 比 Fanout 灵活的代价是"多了一个会写错的东西"。key 写错不报错，直接丢消息，这是后面要重点讲的坑。

路由原理与投递结果

以课程案例的绑定关系为准：direct.queue1 绑定 red、blue；direct.queue2 绑定 red、yellow。四条消息的投递结果如下。

routingKey	direct.queue1（blue, red）	direct.queue2（red, yellow）	结果说明
`red`	收到	收到	两个队列都绑定了 red，等价于一次广播
`blue`	收到	未收到	只有 queue1 绑定 blue
`yellow`	未收到	收到	只有 queue2 绑定 yellow
`green`	未收到	未收到	没有队列绑定 green，消息被交换机丢弃

最后一行是 Direct 模式最容易被忽略的行为：交换机不会因为你发了个陌生 key 而报错，消息直接被丢掉。RabbitMQ 只在 mandatory=true 且配置了 ReturnCallback 时才把无法路由的消息退回给生产者，默认是静默丢弃。

动手前的工程结构

案例沿用 mq-demo 父子工程，publisher 发消息，consumer 收消息。本文涉及的 Direct 相关文件如下。

mq-demo
├── pom.xml                                    # 父工程，Spring Boot 2.x + spring-boot-starter-amqp
├── publisher
│   └── src
│       ├── main/java/cn/itcast/mq/PublisherApplication.java
│       ├── main/resources/application.yml
│       └── test/java/cn/itcast/mq/spring/SpringAmqpTest.java      # Direct 发送测试
└── consumer
    └── src
        ├── main/java/cn/itcast/mq/ConsumerApplication.java
        ├── main/java/cn/itcast/mq/config/FanoutConfig.java         # @Bean 声明式（对照）
        ├── main/java/cn/itcast/mq/listener/SpringRabbitListener.java  # 注解式声明（本文重点）
        └── main/resources/application.yml

两个服务的连接配置一致：

spring:
  rabbitmq:
    host: 192.168.150.101 # RabbitMQ 服务地址
    port: 5672
    username: itcast
    password: 123321
    virtual-host: /

基于注解声明队列与交换机

Fanout 那节课用的是 @Bean 声明式：一个交换机、两个队列、两个 Binding，光声明就写了五个 Bean。Direct 还要额外指定 key，会更啰嗦，所以这里换成注解式——直接在消费者方法上把交换机、队列、绑定关系一次性写完。

cn.itcast.mq.listener.SpringRabbitListener（真实项目代码，Direct 部分）：

package cn.itcast.mq.listener;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.time.LocalTime;
import java.util.Map;

@Component
public class SpringRabbitListener {

    @RabbitListener(queues = "simple.queue")
    public void listenWorkQueue1(String msg) throws InterruptedException {
        System.out.println("消费者1接收到消息：【" + msg + "】" + LocalTime.now());
        Thread.sleep(20);
    }

    @RabbitListener(queues = "simple.queue")
    public void listenWorkQueue2(String msg) throws InterruptedException {
        System.err.println("消费者2........接收到消息：【" + msg + "】" + LocalTime.now());
        Thread.sleep(200);
    }

    // 注解式声明：队列 direct.queue1 绑定 itcast.direct，bindingKey = red / blue
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue1"),
            exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
            key = {"red", "blue"}
    ))
    public void listenDirectQueue1(String msg) {
        System.out.println("消费者接收到direct.queue1的消息：【" + msg + "】");
    }

    // 注解式声明：队列 direct.queue2 绑定同一个交换机，bindingKey = red / yellow
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue2"),
            exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
            key = {"red", "yellow"}
    ))
    public void listenDirectQueue2(String msg) {
        System.out.println("消费者接收到direct.queue2的消息：【" + msg + "】");
    }

    @RabbitListener(queues = "object.queue")
    public void listenObjectQueue(Map<String, Object> msg) {
        System.out.println("接收到object.queue的消息：" + msg);
    }
}

一个容易踩的启动问题：上面 simple.queue 和 object.queue 这两个方法只是按名字订阅，不负责创建队列。simple.queue 在前面简单队列模型一篇里已创建过，object.queue 在消息转换器那篇里也创建过，所以课程环境里直接能跑；但全新环境（管理台里没有这两个队列）启动 consumer 会报 404 NOT_FOUND 并反复重连。补两个最小声明即可，声明是幂等的，队列已存在时不会重复建：

package cn.itcast.mq.config;

import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class InitialQueuesConfig {

    @Bean
    public Queue simpleQueue() {
        return new Queue("simple.queue");   // 构造方法默认 durable=true
    }

    @Bean
    public Queue objectQueue() {
        return new Queue("object.queue");
    }
}

三个注解的分工：

注解	作用	关键属性
`@QueueBinding`	描述"队列 + 交换机 + key"这组绑定关系	`value`（队列）、`exchange`（交换机）、`key`（bindingKey 数组）
`@Queue`	声明队列	`name`、`durable`、`exclusive`、`autoDelete`
`@Exchange`	声明交换机	`name`、`type`（默认就是 `DIRECT`）、`durable`

@Exchange 的 type 默认值已经是 DIRECT，案例里仍然显式写了 ExchangeTypes.DIRECT。建议保持这个习惯：读代码的人不用去查默认值，也不会在复制粘贴去写 Topic 时漏改。

启动 consumer 后打开 RabbitMQ 管理台（15672 端口），可以看到 itcast.direct 交换机下挂了两条绑定：direct.queue1 → red, blue，direct.queue2 → red, yellow。声明动作发生在监听容器启动时，consumer 不起来的话这些资源不会存在。

消息发送

发送侧只多了第二个参数 routingKey，其余和 Fanout 完全一致。

package cn.itcast.mq.spring;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSendDirectExchange() {
        // 交换机名称
        String exchangeName = "itcast.direct";
        // 消息
        String message = "hello, red!";
        // 发送消息，第二个参数就是 routingKey
        rabbitTemplate.convertAndSend(exchangeName, "red", message);
    }
}

想一次验证多条 key，直接循环发：

package cn.itcast.mq.spring;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpDirectBatchTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 依次发送 red / blue / yellow / green，观察 consumer 控制台：
     * red    -> queue1 + queue2 都收到
     * blue   -> 只有 queue1 收到
     * yellow -> 只有 queue2 收到
     * green  -> 无人收到，消息被交换机丢弃
     */
    @Test
    public void testSendDirectExchangeBatch() throws InterruptedException {
        String exchangeName = "itcast.direct";
        String[] keys = {"red", "blue", "yellow", "green"};
        for (String key : keys) {
            String message = "hello, " + key + "!";
            rabbitTemplate.convertAndSend(exchangeName, key, message);
            System.out.println("已发送 routingKey=" + key + " 的消息：" + message);
            Thread.sleep(500);
        }
    }
}

实测结果（Thread.sleep(500) 是为了让控制台输出按 key 分组，方便对照）。注意 publisher 是 JUnit 起的独立 JVM，consumer 是 Spring Boot 起的另一个 JVM，输出在两个终端里分别看，谁先谁后不影响结果——消息发进 RabbitMQ 后由 consumer 异步消费：

publisher 控制台：

已发送 routingKey=red 的消息：hello, red!
已发送 routingKey=blue 的消息：hello, blue!
已发送 routingKey=yellow 的消息：hello, yellow!
已发送 routingKey=green 的消息：hello, green!

consumer 控制台：

消费者接收到direct.queue1的消息：【hello, red!】
消费者接收到direct.queue2的消息：【hello, red!】
消费者接收到direct.queue1的消息：【hello, blue!】
消费者接收到direct.queue2的消息：【hello, yellow!】
（routingKey=green 的消息无任何消费者输出）

两种声明写法对比

对比项	`@Bean` 声明式	`@RabbitListener` 注解式
代码位置	独立的 `@Configuration` 类	消费者方法上方
一个队列绑两个 key	需要两个 `Binding` Bean	`key = {"red", "blue"}` 一行
声明时机	容器刷新时创建 Bean	监听容器启动时解析注解
适合场景	交换机/队列需要被多个服务复用、要精细控制参数	队列只给当前消费者用，快速开发
可读性	声明集中，但类会越写越长	声明与消费逻辑在一起，一眼看清"我消费的是哪个队列、哪个 key"

同样的绑定关系，用 @Bean 写是这样的：

package cn.itcast.mq.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class DirectConfig {

    // itcast.direct
    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange("itcast.direct");
    }

    // direct.queue1
    @Bean
    public Queue directQueue1() {
        return new Queue("direct.queue1");
    }

    // direct.queue1 绑定 red
    @Bean
    public Binding directBinding1Red(Queue directQueue1, DirectExchange directExchange) {
        return BindingBuilder.bind(directQueue1).to(directExchange).with("red");
    }

    // direct.queue1 绑定 blue：一个 key 一个 Binding，不能省
    @Bean
    public Binding directBinding1Blue(Queue directQueue1, DirectExchange directExchange) {
        return BindingBuilder.bind(directQueue1).to(directExchange).with("blue");
    }

    // direct.queue2
    @Bean
    public Queue directQueue2() {
        return new Queue("direct.queue2");
    }

    @Bean
    public Binding directBinding2Red(Queue directQueue2, DirectExchange directExchange) {
        return BindingBuilder.bind(directQueue2).to(directExchange).with("red");
    }

    @Bean
    public Binding directBinding2Yellow(Queue directQueue2, DirectExchange directExchange) {
        return BindingBuilder.bind(directQueue2).to(directExchange).with("yellow");
    }
}

BindingBuilder.bind(queue).to(directExchange).with("red") 里的 with(...) 一次只能给一个 key，所以四个 key 就写四个 Binding。这就是课程里放弃 @Bean 改用注解的直接原因。

典型落地场景：日志分级

日志是 Direct 模式最贴切的用法：同一份日志流，error 要立刻告警，info 只要落归档，warn 两边都不急。

生产者按级别取 routingKey：

package cn.itcast.mq.log;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class LogProducer {

    private static final String LOG_EXCHANGE = "log.direct";

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 发送日志消息，routingKey 就是日志级别
     *
     * @param level   error / warn / info
     * @param content 日志正文
     */
    public void send(String level, String content) {
        rabbitTemplate.convertAndSend(LOG_EXCHANGE, level, content);
    }
}

消费者用注解声明绑定，告警队列只绑 error，归档队列绑三个级别：

package cn.itcast.mq.log;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class LogListener {

    // 告警队列：只关心 error，fatal 级别要加就往 key 数组里补
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "log.alarm.queue"),
            exchange = @Exchange(name = "log.direct", type = ExchangeTypes.DIRECT),
            key = {"error"}
    ))
    public void listenAlarm(String msg) {
        System.err.println("[告警] 触发实时告警：" + msg);
    }

    // 归档队列：全量接收，注意 error 也要绑，否则告警过的日志反而丢了归档
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "log.archive.queue"),
            exchange = @Exchange(name = "log.direct", type = ExchangeTypes.DIRECT),
            key = {"error", "warn", "info"}
    ))
    public void listenArchive(String msg) {
        System.out.println("[归档] 写入存储：" + msg);
    }
}

这里有个容易漏的点：归档队列如果不绑 error，那些触发了告警的 error 日志反而不会进归档，排查问题时就查不到原始日志。绑多个 key 正好是 Direct 的强项。

实战坑

key 不匹配，消息被丢且没有报错。 「red」写成「Red」、多了个空格、写成了「read」，交换机都不会报错，消息直接消失。排查顺序：

管理台打开交换机详情页，核对 Bindings 里的 key 与代码里的 routingKey 是否逐字符一致
命令行核对：rabbitmqctl list_bindings
想让生产者感知到"没路由出去"，开启 mandatory 并注册回调

package cn.itcast.mq.config;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;

@Configuration
public class RabbitReturnConfig {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * mandatory=true：消息无法路由到任何队列时，退回给生产者而不是静默丢弃。
     * 没有这段配置，routingKey 写错就完全没有任何痕迹。
     */
    @PostConstruct
    public void initReturnCallback() {
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnsCallback(returned -> {
            System.err.println("消息未路由成功：exchange=" + returned.getExchange()
                    + ", routingKey=" + returned.getRoutingKey()
                    + ", replyCode=" + returned.getReplyCode()
                    + ", replyText=" + returned.getReplyText());
        });
    }
}

交换机不存在和绑定不匹配是两回事。 向不存在的交换机发消息会触发 channel 级错误 404 NOT_FOUND，Spring 这边抛异常、连接被关掉；而交换机存在但没有匹配 binding，是静默丢弃，什么都不报。看日志里有没有 NOT_FOUND 可以快速区分这两种情况。

一个队列绑多个 key。 注解式写 key = {"red", "blue"}；@Bean 式必须一个 key 一条 Binding，写到一个 with(...) 里编译不过。

Direct 是精确匹配，不是前缀匹配。 把 bindingKey 写成 order.*，只有 routingKey 字面量恰好等于 order.* 才会命中，order.create 匹配不上。要玩通配符得换 TopicExchange，bindingKey 用 order.*（匹配一个词）或 order.#（匹配零到多个词），那是下一篇的内容。

声明是幂等的，但参数冲突会报错。 同名队列用不同参数（比如一边 durable 一边非 durable）重复声明，会抛 PRECONDITION_FAILED - inequivalent arg 'durable'。改队列名或修改参数前，先把旧队列在管理台删掉。

API 速览

API / 注解	说明
`rabbitTemplate.convertAndSend(exchange, routingKey, object)`	向指定交换机发送消息，第二参数是 routingKey；默认用 SimpleMessageConverter 做 JDK 序列化
`@RabbitListener(bindings = @QueueBinding(...))`	消费方法的注解，同时完成交换机、队列、绑定关系的声明
`@QueueBinding`	绑定关系容器，属性 `value`（队列）、`exchange`（交换机）、`key`（bindingKey 数组）
`@Queue(name = "direct.queue1")`	声明队列，`name` 为队列名
`@Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT)`	声明交换机，`type` 默认 `DIRECT`
`ExchangeTypes.DIRECT / TOPIC / FANOUT / HEADERS`	交换机类型枚举，来自 `org.springframework.amqp.core.ExchangeTypes`
`new DirectExchange("itcast.direct")`	`@Bean` 式的交换机声明
`BindingBuilder.bind(queue).to(directExchange).with("red")`	`@Bean` 式的绑定声明，一次只能绑一个 key
`rabbitTemplate.setMandatory(true)` + `setReturnsCallback(...)`	让不可路由的消息退回生产者，避免静默丢消息

官方文档

总结

Direct 的核心就一句话：routingKey 与 bindingKey 全等才投递，比 Fanout 多了一个 key 维度，也因此多了一个会写错的地方
所有队列绑同一个 key 时，Direct 等价于 Fanout；这是它的上限，不是它的典型用法
声明优先用注解式 @QueueBinding，一个队列多个 key 写在一个数组里，@Bean 式则是一个 key 一条 Binding
key 对不上消息会被静默丢弃，生产环境建议开 mandatory + ReturnsCallback
Direct 不支持通配符，order.* 在 Direct 里只是普通字符串，需要模糊匹配就换 TopicExchange

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/Tyro_java/article/details/167364345

Java 中间件之 RabbitMQ 快速入门: SpringAMQP 的 DirectExchange 路由模式

概述

纲要

Direct 与 Fanout 的差异

路由原理与投递结果

动手前的工程结构

基于注解声明队列与交换机

消息发送

两种声明写法对比

典型落地场景：日志分级

实战坑

API 速览

官方文档

总结

评论

评论列表

微信小程序

QQ小程序

关于作者