RabbitMQ 入门指南（五）

古城微笑少年丶 2022-12-26 11:20 135阅读 0赞

## Topic ##

在上一节的示例中，我们使用 direct 类型的交换机，使得消费者可以自由选择接收各种错误级别的日志消息。

但是，direct 这种完全匹配的方式仍然有一定的局限性，无法满足实际的业务场景需要。direct 无法基于多重标准进行路由匹配。

比如，假设我们既要根据日志消息的错误级别来订阅消息，又要根据日志消息的设备来源来订阅消息。

为了实现这一功能，就需要用到更为灵活的交换机。

下面我们来介绍 topic 类型的交换机。

## topic exchange ##

topic 交换机与 direct 交换机类似，也是将消息路由给 binding\_key 与 routing\_key 相匹配的队列。 只不过，topic 交换机不是完全匹配，它的匹配规则更加灵活，支持模糊匹配。

**topic 类型的交换机有如下约定：**

*  routing\_key 是由 **.** 号分隔的多个单词组成的字符串，长度不得超过 255 字节，如 `quick.orange.rabbit`。
 *  上述约定同样适用于 binding\_key。
 *  binding\_key 还支持两个特殊字符 **\*** 和 **\#**，用于模糊匹配。

**\*（星号）只能匹配一个单词。**

**\#（井号）可以匹配零个或多个单词。**

![image][]

在这个例子中，我们发送消息来描述动物。

routing\_key 包含三个单词和两个点号，三个单词分别描述动物的速度、颜色和种类。

我们进行三次绑定。

*  Q1 的 binding\_key 为 `*.orange.*`
 *  Q2 的 binding\_key 为 `*.*.rabbit`
 *  Q2 的 binding\_key 为 `lazy.#`

上述绑定关系可以概括为：

*  Q1 对所有颜色为 orange 的动物感兴趣
 *  Q2 对兔子和速度为 lazy 的动物感兴趣。

这样绑定后：

*  routing\_key 为 `quick.orange.rabbit` 的消息，会被发送给 Q1 和 Q2。
 *  routing\_key 为 `lazy.orange.elephant` 的消息，也会被发送给 Q1 和 Q2。
 *  routing\_key 为 `quick.orange.fox` 的消息，仅被发送给 Q1。
 *  routing\_key 为 `lazy.brown.fox` 的消息，仅被发送给 Q2。
 *  routing\_key 为 `lazy.pink.rabbit` 的消息，仅被发送给 Q2，且只发送一次，即使它匹配了两个绑定。
 *  routing\_key 为 `quick.brown.fox` 的消息，没有匹配到任何绑定，故会被丢弃（不会发送给队列）。
 *  routing\_key 为 `lazy.orange.male.rabbit` 的消息，仅被发送给 Q2。

**关于 topic exchange 的特殊说明：**

*  如果一个队列的 binding\_key 是 **\#**，它将接收与之绑定的交换机中所有的消息，无论生产消息时的 routing\_key 是什么。类似于 fanout 类型的交换机。
 *  如果 binding\_key 中没有特殊字符 **\*** 和 **\#**，那么，topic exchange 的行为表现和 direct exchange 一样。

## 最终的代码示例 ##

emit\_log\_topic.php，作为生产者，内容如下：

<?php
    
    require_once __DIR__ . '/vendor/autoload.php';
    use PhpAmqpLib\Connection\AMQPStreamConnection;
    use PhpAmqpLib\Message\AMQPMessage;
    
    $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
    $channel = $connection->channel();
    
    $channel->exchange_declare('topic_logs', 'topic', false, false, false);
    
    $routing_key = isset($argv[1]) && !empty($argv[1]) ? $argv[1] : 'anonymous.info';
    $data = implode(' ', array_slice($argv, 2));
    if (empty($data)) {
        $data = "Hello World!";
    }
    
    $msg = new AMQPMessage($data);
    
    $channel->basic_publish($msg, 'topic_logs', $routing_key);
    
    echo ' [x] Sent ', $routing_key, ':', $data, "\n";
    
    $channel->close();
    $connection->close();

receive\_logs\_topic.php，作为消费者，内容如下：

<?php
    
    require_once __DIR__ . '/vendor/autoload.php';
    use PhpAmqpLib\Connection\AMQPStreamConnection;
    
    $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
    $channel = $connection->channel();
    
    $channel->exchange_declare('topic_logs', 'topic', false, false, false);
    
    list($queue_name, ,) = $channel->queue_declare("", false, false, true, false);
    
    $binding_keys = array_slice($argv, 1);
    if (empty($binding_keys)) {
        file_put_contents('php://stderr', "Usage: $argv[0] [binding_key]\n");
        exit(1);
    }
    
    foreach ($binding_keys as $binding_key) {
        $channel->queue_bind($queue_name, 'topic_logs', $binding_key);
    }
    
    echo " [*] Waiting for logs. To exit press CTRL+C\n";
    
    $callback = function ($msg) {
        echo ' [x] ', $msg->delivery_info['routing_key'], ':', $msg->body, "\n";
    };
    
    $channel->basic_consume($queue_name, '', false, true, false, false, $callback);
    
    while ($channel->is_consuming()) {
        $channel->wait();
    }
    
    $channel->close();
    $connection->close();

运行三个消费者 C1、C2 和 C3。

# C1，接收所有的日志消息
    php receive_logs_topic.php "#"

# C2，接收设备来源是 kern 的日志消息
    php receive_logs_topic.php "kern.*"

# C3，接收错误级别是 critical 的日志消息
    php receive_logs_topic.php "*.critical"

运行一个生产者 P

# P，发送消息
    php emit_log_topic.php "kern.critical" "A critical kernel error"

**参考文献**

\[1\] [https://www.rabbitmq.com/tutorials/tutorial-five-php.html][https_www.rabbitmq.com_tutorials_tutorial-five-php.html]

[image]: /images/20221120/c59387ed80bc4761b3ff1e7c570abb20.png
[https_www.rabbitmq.com_tutorials_tutorial-five-php.html]: https://www.rabbitmq.com/tutorials/tutorial-five-php.html