RabbitMQ 消息模式实战：从简单队列到复杂路由（三）

精准投递：路由模式

路由模式详解

路由模式是 RabbitMQ 中一种功能强大且灵活的消息传递模式，它在发布订阅模式的基础上，引入了路由键（Routing Key）的概念，实现了消息的精准路由和分发 。在路由模式中，交换机同样扮演着关键角色，不过与发布订阅模式中使用的扇形交换机不同，路由模式通常使用直连交换机（Direct Exchange） 。

直连交换机的工作原理是基于路由键的精确匹配。当生产者将消息发送到直连交换机时，必须指定一个路由键，这个路由键就像是一个地址标签，标识着消息的目的地 。交换机在接收到消息后，会根据消息的路由键，在与自己绑定的队列中寻找匹配的队列。只有那些绑定键（Binding Key，队列与交换机绑定时指定的键）与消息路由键完全相同的队列，才会接收到该消息 。例如，假设有一个订单处理系统，有两个队列，分别是 “new_order_queue”（新订单队列）和 “urgent_order_queue”（紧急订单队列），与直连交换机绑定，绑定键分别为 “new_order” 和 “urgent_order” 。当生产者发送一条消息，路由键为 “new_order” 时，直连交换机就会将这条消息精准地路由到 “new_order_queue” 队列中；如果路由键为 “urgent_order”，则会被路由到 “urgent_order_queue” 队列 。这种精确匹配的机制，确保了消息能够被准确地发送到目标队列，满足了不同业务场景下对消息定向处理的需求 。

应用场景分析

路由模式在实际业务中有着广泛的应用场景。以日志处理系统为例，系统运行过程中会产生各种不同级别的日志，如错误日志（error）、信息日志（info）、警告日志（warning）等 。通过路由模式，可以将不同级别的日志消息发送到不同的队列进行处理 。将错误日志消息的路由键设置为 “error”，与绑定键为 “error” 的错误日志队列绑定，当错误日志产生时，就会被精准地路由到该队列，由专门的错误处理模块进行处理，例如将错误信息记录到数据库、发送告警通知等 ；而信息日志消息的路由键设置为 “info”，路由到信息日志队列，可能进行简单的记录和统计分析 。这样，通过路由模式，能够对不同类型的日志进行分类处理，提高日志处理的效率和针对性 。

在电商系统的订单分类处理场景中，路由模式也发挥着重要作用 。订单可以根据不同的属性进行分类，如普通订单、团购订单、跨境订单等 。将普通订单消息的路由键设置为 “normal_order”，与绑定键为 “normal_order” 的普通订单队列绑定；团购订单消息的路由键设置为 “group_buy_order”，与团购订单队列绑定 。当订单生成时，根据订单类型设置相应的路由键，订单消息就会被准确地路由到对应的队列，由不同的订单处理模块进行处理，比如普通订单按照常规流程进行处理，团购订单可能需要进行团购规则的校验、团购人数统计等特殊处理 。通过这种方式，实现了订单的分类管理和高效处理 。

代码实战演示

下面通过 Java 代码来演示如何实现路由模式。假设我们有一个订单处理系统，根据订单类型将订单消息发送到不同的队列。

首先，在 Maven 项目的pom.xml文件中添加 RabbitMQ 客户端依赖：

<dependency>

<groupId>com.rabbitmq</groupId>

<artifactId>amqp-client</artifactId>

<version>5.16.0</version>

</dependency>

生产者代码如下：

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.ConnectionFactory;

public class OrderProducer {

private final static String EXCHANGE_NAME = "order_exchange";

private final static String NORMAL_ORDER_ROUTING_KEY = "normal_order";

private final static String GROUP_BUY_ORDER_ROUTING_KEY = "group_buy_order";

public static void main(String[] argv) throws Exception {

ConnectionFactory factory = new ConnectionFactory();

factory.setHost("localhost");

try (Connection connection = factory.newConnection();

Channel channel = connection.createChannel()) {

// 声明直连交换机

channel.exchangeDeclare(EXCHANGE_NAME, "direct");

// 模拟发送普通订单消息

String normalOrderMessage = "Normal Order: Product A, Quantity 1";

channel.basicPublish(EXCHANGE_NAME, NORMAL_ORDER_ROUTING_KEY, null, normalOrderMessage.getBytes("UTF-8"));

System.out.println(" [x] Sent '" + normalOrderMessage + "' with routing key: " + NORMAL_ORDER_ROUTING_KEY);

// 模拟发送团购订单消息

String groupBuyOrderMessage = "Group Buy Order: Product B, Quantity 10";

channel.basicPublish(EXCHANGE_NAME, GROUP_BUY_ORDER_ROUTING_KEY, null, groupBuyOrderMessage.getBytes("UTF-8"));

System.out.println(" [x] Sent '" + groupBuyOrderMessage + "' with routing key: " + GROUP_BUY_ORDER_ROUTING_KEY);

}

}

}

在生产者代码中，首先创建了连接工厂、连接和通道。然后使用channel.exchangeDeclare方法声明了一个名为 “order_exchange” 的直连交换机。接着，分别模拟发送了普通订单消息和团购订单消息，发送时指定了对应的路由键 。

消费者代码（以普通订单消费者为例）：

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.ConnectionFactory;

import com.rabbitmq.client.DeliverCallback;

public class NormalOrderConsumer {

private final static String EXCHANGE_NAME = "order_exchange";

private final static String NORMAL_ORDER_ROUTING_KEY = "normal_order";

private final static String QUEUE_NAME = "normal_order_queue";

public static void main(String[] argv) throws Exception {

ConnectionFactory factory = new ConnectionFactory();

factory.setHost("localhost");

try (Connection connection = factory.newConnection();

Channel channel = connection.createChannel()) {

// 声明直连交换机

channel.exchangeDeclare(EXCHANGE_NAME, "direct");

// 声明队列

channel.queueDeclare(QUEUE_NAME, false, false, false, null);

// 将队列绑定到交换机，并指定绑定键

channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, NORMAL_ORDER_ROUTING_KEY);

System.out.println(" [*] Waiting for normal order messages. To exit press CTRL+C");

DeliverCallback deliverCallback = (consumerTag, delivery) -> {

String message = new String(delivery.getBody(), "UTF-8");

System.out.println(" [x] Received '" + message + "'");

};

channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });

}

}

}

普通订单消费者代码中，同样先声明了直连交换机，然后声明了 “normal_order_queue” 队列，并使用channel.queueBind方法将队列绑定到交换机上，指定绑定键为 “normal_order” 。定义了消息处理回调函数DeliverCallback，当接收到消息时，会打印出消息内容 。最后使用channel.basicConsume方法开始消费消息 。通过运行生产者和消费者代码，可以看到订单消息会根据路由键被准确地发送到对应的队列并被处理，成功实现了路由模式 。

灵活匹配：主题模式

主题模式原理剖析

主题模式是 RabbitMQ 中一种功能强大且灵活的消息路由模式，它在路由模式的基础上，进一步拓展了路由规则，通过引入通配符来实现更灵活的消息匹配和分发 。在主题模式中，同样使用交换机来接收生产者发送的消息，并根据消息的路由键将其路由到匹配的队列中 。

主题模式使用的交换机类型是主题交换机（Topic Exchange） 。与直连交换机不同，主题交换机在匹配路由键时，支持使用通配符进行模糊匹配，从而大大提高了消息路由的灵活性 。主题模式中支持两种通配符：“”（星号）和 “#”（井号） 。“” 代表任意一个单词，“#” 代表零个或多个单词 。这里所说的单词，是指由 “.” 分隔的字符串片段 。例如，路由键 “order.new” 中包含两个单词 “order” 和 “new” 。

当生产者发送消息时，会指定一个路由键，如 “user.create.success” 。交换机在接收到消息后，会将该路由键与各个队列的绑定键进行匹配 。如果某个队列的绑定键为 “user.#”，那么该队列将匹配到这条消息，因为 “#” 可以匹配零个或多个单词，这里 “user.create.success” 中的 “create.success” 被 “#” 匹配 ；如果绑定键为 “user..success”，同样也能匹配到，因为 “” 匹配了 “create” 这一个单词 。通过这种通配符的匹配方式，一条消息可以被路由到多个符合条件的队列，实现了消息的灵活分发 。

实际应用场景

主题模式在实际业务中有广泛的应用场景。以内容推送系统为例，假设一个内容平台有多种类型的内容，如新闻、视频、文章等，同时有不同兴趣偏好的用户群体 。通过主题模式，可以将内容发布的消息进行灵活路由 。将新闻相关的内容消息的路由键设置为 “news.*”，如 “news.politics”（政治新闻）、“news.entertainment”（娱乐新闻） 。对于关注政治新闻的用户队列，绑定键设置为 “news.politics”，这样当有新的政治新闻发布时，该用户队列就能接收到消息，从而将新闻推送给关注政治的用户 ；对于关注所有新闻的用户队列，绑定键设置为 “news.#”，则可以接收所有类型的新闻消息 。通过这种方式，实现了根据用户兴趣精准推送内容，提高了用户体验和内容传播的效率 。

在消息过滤场景中，主题模式也发挥着重要作用 。比如在一个分布式系统中，各个模块会产生大量的日志消息 。为了对这些日志进行有效的管理和分析，可以使用主题模式进行消息过滤 。将错误日志消息的路由键设置为 “error.*”，如 “error.database”（数据库错误）、“error.network”（网络错误） 。创建一个专门处理数据库错误的队列，绑定键设置为 “error.database”，当有数据库错误日志产生时，该队列就能接收到消息，进行针对性的处理，如记录详细的错误信息、发送告警通知给数据库管理员等 ；对于记录所有错误日志的队列，绑定键设置为 “error.#”，可以收集所有类型的错误日志，便于后续的统计和分析 。通过主题模式的灵活匹配，实现了对日志消息的高效过滤和处理 。

代码示例讲解

下面通过 Java 代码来展示如何使用主题模式实现消息的灵活分发。假设我们有一个内容推送系统，根据不同的内容类型和用户兴趣进行消息路由 。

首先，在 Maven 项目的pom.xml文件中添加 RabbitMQ 客户端依赖：

<dependency>

<groupId>com.rabbitmq</groupId>

<artifactId>amqp-client</artifactId>

<version>5.16.0</version>

</dependency>

生产者代码如下：

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.ConnectionFactory;

public class ContentProducer {

private final static String EXCHANGE_NAME = "content_exchange";

public static void main(String[] argv) throws Exception {

ConnectionFactory factory = new ConnectionFactory();

factory.setHost("localhost");

try (Connection connection = factory.newConnection();

Channel channel = connection.createChannel()) {

// 声明主题交换机

channel.exchangeDeclare(EXCHANGE_NAME, "topic");

// 模拟发布新闻消息

String newsRoutingKey = "news.politics";

String newsMessage = "New Political News: New Policy Announcement";

channel.basicPublish(EXCHANGE_NAME, newsRoutingKey, null, newsMessage.getBytes("UTF-8"));

System.out.println(" [x] Sent '" + newsMessage + "' with routing key: " + newsRoutingKey);

// 模拟发布视频消息

String videoRoutingKey = "video.entertainment";

String videoMessage = "New Entertainment Video: Popular Movie Trailer";

channel.basicPublish(EXCHANGE_NAME, videoRoutingKey, null, videoMessage.getBytes("UTF-8"));

System.out.println(" [x] Sent '" + videoMessage + "' with routing key: " + videoRoutingKey);

}

}

}

在生产者代码中，创建了连接工厂、连接和通道后，使用channel.exchangeDeclare方法声明了一个名为 “content_exchange” 的主题交换机 。然后分别模拟发布了新闻消息和视频消息，发送时指定了对应的路由键 。

消费者代码（以关注政治新闻的用户消费者为例）：

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.ConnectionFactory;

import com.rabbitmq.client.DeliverCallback;

public class PoliticsNewsConsumer {

private final static String EXCHANGE_NAME = "content_exchange";

private final static String QUEUE_NAME = "politics_news_queue";

private final static String BINDING_KEY = "news.politics";

public static void main(String[] argv) throws Exception {

ConnectionFactory factory = new ConnectionFactory();

factory.setHost("localhost");

try (Connection connection = factory.newConnection();

Channel channel = connection.createChannel()) {

// 声明主题交换机

channel.exchangeDeclare(EXCHANGE_NAME, "topic");

// 声明队列

channel.queueDeclare(QUEUE_NAME, false, false, false, null);

// 将队列绑定到交换机，并指定绑定键

channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, BINDING_KEY);

System.out.println(" [*] Waiting for political news messages. To exit press CTRL+C");

DeliverCallback deliverCallback = (consumerTag, delivery) -> {

String message = new String(delivery.getBody(), "UTF-8");

System.out.println(" [x] Received '" + message + "'");

};

channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });

}

}

}

关注政治新闻的用户消费者代码中，先声明了主题交换机和 “politics_news_queue” 队列，然后使用channel.queueBind方法将队列绑定到交换机上，指定绑定键为 “news.politics” 。定义了消息处理回调函数DeliverCallback，当接收到消息时，会打印出消息内容 。最后使用channel.basicConsume方法开始消费消息 。通过运行生产者和消费者代码，可以看到消息会根据路由键和绑定键的匹配规则，被准确地路由到对应的队列并被处理，成功实现了主题模式下消息的灵活分发 。