RabbitMQ的Direct模式：消息队列的路由运用

2026-07-16 10:59小蒋

背景：从广播到精准路由

在了解了RabbitMQ的Fanout（广播）模式后，我们知道了如何将一条消息同时发送给所有感兴趣的消费者。然而，在实际业务场景中，我们往往需要更精细的消息分发策略——不是所有消息都需要被所有消费者接收，不同的消费者应该只关心自己感兴趣的那一类消息。

例如，在一个日志处理系统中：

错误日志需要被发送到报警服务，立即通知运维人员
信息日志可能只需要被写入存储，用于后续审计
调试日志则可能只对开发环境开放

如果使用Fanout模式，所有消费者都会收到每一条日志消息，这显然会造成资源浪费和处理混乱。此时，我们需要一种能够根据消息的“路由键”（Routing Key）进行有选择性分发的模式——这就是RabbitMQ的Direct（直连）模式。

Direct模式的核心思想是：生产者发送消息时指定一个路由键，交换机根据这个路由键将消息精确地投递到绑定了相同路由键的队列中。它就像一套高效的邮件分拣系统，根据信封上的“收件人地址”将邮件准确送达对应的收件箱。

核心原理：Direct模式的工作流程

在Direct模式下，消息的流转遵循更精确的路径：

生产者：发送消息到Direct类型的交换机，并必须指定一个明确的路由键（Routing Key）。
交换机（Direct类型）：收到消息后，会精确匹配路由键，将消息投递到与该路由键绑定的队列。
队列与绑定：每个队列在绑定时，会指定一个它感兴趣的路由键。只有当消息的路由键与队列绑定的路由键完全一致时，消息才会进入该队列。
消费者：从各自绑定的队列中获取消息。不同消费者可以监听不同路由键，实现消息的按需分发。

因此，Direct模式非常适合用于按类型、级别或来源进行消息分类处理的场景，如日志分级、任务分配、事件路由等。

代码实践：用PHP实现Direct模式

在开始之前，请确保你的PHP项目已经通过Composer安装了RabbitMQ扩展（php-amqplib/php-amqplib）。

1. 生产者：publish.php

生产者负责创建Direct交换机，并发送带有路由键的消息。

<?php
require_once __DIR__ . '/vendor/autoload.php';

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

// 连接配置（请替换为你的实际信息）
$dbName = 'sanqing';
$dbPwd = '111111';
$tableName = 'order';

$connection = new AMQPStreamConnection('localhost', 5672, $dbName, $dbPwd, $tableName);
$channel = $connection->channel();

// 声明一个Direct类型的交换机，名为 'direct_log'
$exc_name = 'direct_log';
$channel->exchange_declare($exc_name, 'direct', false, false, false);

// 定义路由键（可以修改为 'info' 或 'error' 测试不同效果）
$routing_key = 'info';

// 发送消息到交换机，并携带路由键
$data = 'this is ' . $routing_key . ' direct message';
$msg = new AMQPMessage($data, ['delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT]);
$channel->basic_publish($msg, $exc_name, $routing_key);

echo " [x] 消息已发送，路由键为：", $routing_key, "\n";

$channel->close();
$connection->close();

2. 消费者：info.php 和 error.php

消费者负责创建临时队列，并将队列绑定到交换机时指定感兴趣的路由键，然后接收消息。你可以复制并修改路由键来创建不同的消费者。

消费者 info.php（监听 `info` 路由）

<?php
require_once __DIR__ . '/vendor/autoload.php';

use PhpAmqpLib\Connection\AMQPStreamConnection;

// 连接配置（请与生产者保持一致）
$dbName = 'sanqing';
$dbPwd = '111111';
$tableName = 'order';

$connection = new AMQPStreamConnection('localhost', 5672, $dbName, $dbPwd, $tableName);
$channel = $connection->channel();

// 声明要绑定的Direct交换机
$exc_name = 'direct_log';
$channel->exchange_declare($exc_name, 'direct', false, false, false);

// 指定此消费者关心的路由键
$routing_key = 'info';

// 创建一个临时队列
list($queue_name, ,) = $channel->queue_declare('', false, false, true, false);

// 将队列绑定到交换机，并指定路由键
$channel->queue_bind($queue_name, $exc_name, $routing_key);

echo " [*] 等待 '{$routing_key}' 消息。要退出请按 CTRL+C\n";

$callback = function ($msg) {
    echo ' [x] 收到消息：', $msg->body, "\n";
    $msg->ack();
};

$channel->basic_qos(null, 1, null);
$channel->basic_consume($queue_name, '', false, false, false, false, $callback);

while ($channel->is_open()) {
    $channel->wait();
}

$channel->close();
$connection->close();

消费者 error.php（监听 `error` 路由）

复制 info.php 文件为 error.php，然后修改其中的 $routing_key = 'error';，其他部分完全相同。这样，error.php 消费者就只会接收路由键为 error 的消息。

3. 运行测试

打开两个终端窗口，分别运行：
bash
```
php info.php
```
和
bash
```
php error.php
```
再打开一个终端窗口，运行生产者（发送 info 消息）：
bash
```
php publish.php
```
观察输出，你会发现只有 info.php 收到了消息，error.php 没有任何输出。
将 publish.php 中的 $routing_key = 'error';，再次执行：
bash
```
php publish.php
```
此时，只有 error.php 收到了消息。

这就是Direct模式的精确路由能力——消息只会被发送到绑定了匹配路由键的队列。

总结

通过本次实践，我们掌握了RabbitMQ中Direct（直连）模式的核心概念与用法。

关键点	说明
核心思想	根据路由键进行精确匹配，实现消息的按需分发。
适用场景	适用于需要根据类型、级别或来源对消息进行分类处理的场景，如日志分级（info/error）、不同业务模块的消息路由等。
关键代码	生产者：`exchange_declare` 设置交换机类型为 `direct`，并在 `basic_publish` 时指定 `routing_key`。消费者：在 `queue_bind` 时指定与生产者对应的 `routing_key`，以接收特定消息。

Direct模式提供了比Fanout模式更精细的控制能力，是构建灵活、可扩展消息系统的基石。理解并掌握它，可以帮助我们更好地设计基于事件驱动的应用架构。

曝光574浏览39