Java-208 RabbitMQ Topic 主题交换器详解：routingKey/bindingKey 通配符与 Java 示例

TL;DR

• 场景：需要按“多维标签”路由消息（如来源+级别/区域+类型）。
• 结论：topic 通过 routingKey（点分词）与 bindingKey（*/#）模式匹配实现精细分发，不匹配会直接丢弃。
• 产出：给出匹配规则、命名规范与 Java Producer/Consumer 代码落地要点+常见故障速查。

RabbitMQ 主题模式

使用 topic 类型的交换器时，队列通过 bindingKey 绑定到交换器，其中 bindingKey 可以包含通配符进行灵活匹配。交换器在路由消息时，会将消息的 routingKey 与队列的 bindingKey 进行模式匹配，从而实现更细粒度的消息分发。

具体来说，bindingKey 支持两种通配符：

在上一个 direct 交换器的模式中，我们只能根据单一的日志级别（如 error、info、warning）来路由消息。如果需要同时考虑日志来源和日志级别两个维度，topic 模式就能很好地满足需求。

应用场景示例： 假设我们有如下需求：

对应的绑定关系可以这样设置：

• 队列1绑定键：cron.error
• 队列2绑定键：kern.*
• 队列3绑定键：*.critical

当消息的 routingKey 为：

• cron.error：会被路由到队列1
• kern.info：会被路由到队列2
• auth.critical：会被路由到队列3

这种模式特别适合需要多维度分类消息的场景，比如：

通过合理设计 routingKey 的命名规则和 bindingKey 的匹配模式，可以实现非常灵活的消息路由机制。

在使用 RabbitMQ 的 topic 类型交换器时，RoutingKey 的设计需要遵循特定的格式规范：

• 必须采用点分单词的形式（dotted-word notation）
• 每个单词之间用点号(.)分隔
• 单词可以是任意有效的字符串
• 整个字符串长度不能超过 255 字节

• 建议使用有意义的业务特征作为单词
• 常见命名模式示例：
  • 金融领域：stock.usd.nyse（股票.货币.交易所）
  • 汽车行业：nyse.vwm（交易所.汽车品牌）
  • 通用示例：quick.orange.rabbit（速度.颜色.动物）

• 基础匹配：队列的 bindingKey 必须与消息的 routingKey 完全匹配才能接收消息
• 特殊通配符：
  • 星号(*)：精确匹配一个单词
    • 示例：*.stock.# 可以匹配 usd.stock 或 eur.stock.nyse
  • 井号(#)：匹配零个或多个单词
    • 示例：stock.# 可以匹配 stock、stock.usd 或 stock.usd.nyse

• 多维度消息路由：比如订单系统可以用 region.product.status 的形式（如 eu.cloth.shipped）
• 层级分类：company.department.role（如 acme.sales.manager）
• 事件溯源：system.component.event（如 inventory.item.restocked）

注意事项：

• 通配符必须作为独立的单词出现，不能嵌入单词中间
• 匹配是大小写敏感的
• 空字符串不能作为 routingKey
• 点号不能出现在开头或结尾

上图中，我们发送消息，消息发送的时候指定 routingKey 包含了三个词，两个点。第一个单词表示动物的速度、第二个是颜色、第三个是物种：speed.color.species。

在 RabbitMQ 的消息路由机制中，topic 类型的交换器（Exchange）提供了灵活的消息路由方式。下面详细说明三种绑定的创建及其匹配规则：

Topic 交换器的行为类比

• 与 Fanout 交换器的相似性

  • 如果在 topic 交换器中绑定键（bindingKey）使用 #（如 # 或 lazy.#），则交换器会广播所有消息到匹配的队列，类似于 fanout 交换器的行为（即忽略 routingKey，向所有绑定队列发送消息）。

• 与 Direct 交换器的相似性

  • 如果在 topic 交换器中绑定键完全不使用 * 或 #（如 orange.rabbit），则交换器会精确匹配 routingKey，类似于 direct 交换器的行为（即仅发送给 routingKey 完全匹配的队列）。

示例场景

假设有以下消息发布：

• quick.orange.rabbit → 匹配 Q2（*.*.rabbit）
• lazy.orange.elephant → 匹配 Q1（*.orange.*）和 Q2（lazy.#）
• orange.rabbit → 不匹配任何绑定，消息被丢弃
• lazy.brown.rabbit → 匹配 Q2（*.*.rabbit 和 lazy.#）

通过这种灵活的绑定方式，topic 交换器可以实现精细化的消息路由策略。

EmitLogTopic

package icu.wzk;
/**
* EmitLogTopic：发布端（Producer）
*
* 目标：
* – 声明 topic 类型交换机 topic_logs
* – 构造 routingKey = speed.color.species（例如 quick.red.rabbit）
* – 发布若干条消息
*
* topic 语义：
* – routingKey 是用 "." 分隔的多个单词（tokens）
* – bindingKey 支持通配符：
* * : 匹配 1 个 token（exactly one word）
* # : 匹配 0..N 个 token（zero or more words）
*
* 示例：
* – bindingKey = "*.red.*" => 匹配 quick.red.rabbit / lazy.red.dog 等（正好 3 段）
* – bindingKey = "quick.#" => 匹配 quick.red.rabbit / quick.orange.cat / quick.xxx.yyy.zzz 等
* – bindingKey = "#.rabbit" => 匹配 lazy.black.rabbit / rabbit（若只有1段也能匹配）/ quick.red.rabbit
*/
public class EmitLogTopic {

private static final String EXCHANGE_NAME = "topic_logs";

private static final String[] SPEED = {"lazy", "quick", "normal"};
private static final String[] COLOR = {"black", "orange", "red", "yellow", "blue", "white", "pink"};
private static final String[] SPECIES = {"dog", "rabbit", "chicken", "horse", "bear", "cat"};

private static final Random RANDOM = new Random();

public static void main(String[] args) throws IOException, TimeoutException {
// 1) 配置连接
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setVirtualHost("/");
factory.setUsername("admin");
factory.setPassword("secret");
factory.setPort(5672);

// 2) 建立连接与 Channel（发完即关闭）
try (Connection connection = factory.newConnection();
Channel channel = connection.createChannel()) {

/**
* 3) 声明 topic 交换机
* topic 交换机用于“多维标签路由”，靠 routingKey 的多段 token 来匹配 bindingKey 规则
*/
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);

// 4) 循环发送 10 条随机消息
for (int i = 0; i < 10; i++) {
String speed = pick(SPEED);
String color = pick(COLOR);
String species = pick(SPECIES);

// 消息体：业务内容（这里用 "-" 连接，纯展示）
String message = speed + "-" + color + "-" + species;

// routingKey：topic 的路由关键（必须用 "." 分段）
String routingKey = speed + "." + color + "." + species;

/**
* 5) 发布消息
* routingKey 会被 topic exchange 用来和队列绑定时的 bindingKey 做模式匹配
*/
channel.basicPublish(
EXCHANGE_NAME,
routingKey,
null,
message.getBytes(StandardCharsets.UTF_8)
);

// 逐条打印，避免只打印最后一条造成误解
System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'");
}
}
}

/** 从数组中随机取一个元素 */
private static String pick(String[] arr) {
return arr[RANDOM.nextInt(arr.length)];
}
}

package icu.wzk;
/**
* Topic 模式消费者示例：ReceiveLogsTopic
*
* 核心逻辑：
* 1) 连接到 RabbitMQ
* 2) 声明一个 topic exchange（topic_logs）
* 3) 创建一个临时队列（server-named / exclusive / auto-delete）
* 4) 使用 bindingKey "*.*.rabbit" 绑定队列到 exchange
* 5) 开始消费，打印匹配到的消息
*
* Topic 匹配规则（重点）：
* – routingKey 用 '.' 分段（比如 "a.b.rabbit"）
* – '*' 匹配“恰好一个分段”
* – '#' 匹配“零个或多个分段”
*
* 所以 "*.*.rabbit" 能匹配：
* – "a.b.rabbit" ✅
* – "x.y.rabbit" ✅
* – "a.rabbit" ❌（只有 2 段）
* – "a.b.c.rabbit" ❌（4 段）
*/
public class ReceiveLogsTopic {

private static final String EXCHANGE_NAME = "topic_logs";

public static void main(String[] args) throws IOException, TimeoutException {

// 1) 创建连接工厂：配置 RabbitMQ 连接参数
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost"); // RabbitMQ 节点地址/域名
factory.setPort(5672); // AMQP 默认端口
factory.setVirtualHost("/"); // vhost：相当于逻辑隔离的命名空间
factory.setUsername("admin"); // 用户名
factory.setPassword("secret"); // 密码

/**
* 2) 建立连接 + 创建 Channel
* 建议使用 try-with-resources，避免进程退出前忘记关闭资源导致泄漏。
* 这里为了保持“常驻消费”，不 close，让进程一直阻塞等待消息。
*/
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

/**
* 3) 声明 topic exchange
* – 如果 exchange 不存在则创建
* – 如果存在则校验类型/参数一致
*/
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);

/**
* 4) 声明一个临时队列（不传队列名）
* queueDeclare() 不带参数时，RabbitMQ 会返回一个随机队列名：
* – exclusive：仅当前连接可用
* – auto-delete：连接断开后队列自动删除（典型的“订阅者临时队列”模式）
*/
String queueName = channel.queueDeclare().getQueue();

/**
* 5) 绑定队列到 exchange，并设置 bindingKey
* bindingKey = "*.*.rabbit"
* 只有 routingKey 符合 topic 匹配规则的消息才会路由到这个队列
*/
String bindingKey = "*.*.rabbit";
channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);

/**
* 6) 定义收到消息后的回调
* message.getBody() 是 byte[]，用 UTF-8 解码成字符串
*/
DeliverCallback callback = (consumerTag, delivery) -> {
String body = new String(delivery.getBody(), StandardCharsets.UTF_8);
System.out.println(bindingKey + " 匹配到的消息：" + body);

// delivery.getEnvelope().getRoutingKey() 可以看到实际 routingKey（常用排查）
// System.out.println("routingKey=" + delivery.getEnvelope().getRoutingKey());
};

/**
* 7) 开始消费
*
* autoAck = true：
* – RabbitMQ 一投递就认为你“已确认”
* – 消费者如果打印前崩溃，消息也不会重新投递（有丢消息风险）
*
* 更可靠的方式：
* – autoAck=false
* – 处理成功后 channel.basicAck(deliveryTag, false)
* – 失败则 basicNack / basicReject
*/
boolean autoAck = true;
channel.basicConsume(queueName, autoAck, callback, consumerTag -> {});

// 程序不退出：保持常驻消费（此处不需要写 while(true)；连接/线程会阻塞等待投递）
}
}

错误速查

其他系列

🚀 AI篇持续更新中（长期更新）

AI炼丹日志-29 – 字节跳动 DeerFlow 深度研究框斜体样式架 私有部署 测试上手 架构研究，持续打造实用AI工具指南！ AI研究-132 Java 生态前沿 2025：Spring、Quarkus、GraalVM、CRaC 与云原生落地 🔗 AI模块直达链接

💻 Java篇持续更新中（长期更新）

Java-207 RabbitMQ Direct 交换器路由：RoutingKey 精确匹配、队列多绑定与日志分流实战 MyBatis 已完结，Spring 已完结，Nginx已完结，Tomcat已完结，分布式服务已完结，Dubbo已完结，MySQL已完结，MongoDB已完结，Neo4j已完结，FastDFS 已完结，OSS已完结，GuavaCache已完结，EVCache已完结，RabbitMQ正在更新… 深入浅出助你打牢基础！ 🔗 Java模块直达链接

📊 大数据板块已完成多项干货更新（300篇）：

包括 Hadoop、Hive、Kafka、Flink、ClickHouse、Elasticsearch 等二十余项核心组件，覆盖离线+实时数仓全栈！ 大数据-278 Spark MLib – 基础介绍 机器学习算法 梯度提升树 GBDT案例 详解 🔗 大数据模块直达链接