RabbitMQ 路由模式的使用

路由模式與發佈訂閱相似，區別是發佈訂閱交換機會將全部消息廣播給全部消費者綁定的隊列，消費者消費所有消息；而路由模式交換機只會將與路由鍵徹底匹配的消息發送給消費者綁定的隊列，消費者只消費該類消息。

安裝依賴

# composer.json
{
    "require": {
        "php-amqplib/php-amqplib": ">=3.0"
    }
}
> composer.phar install

模式結構

路由模式使用的是direct 交換機，這個交換機的路由算法也很簡單，將它收到的全部消息進入綁定鍵與消息的路由鍵徹底匹配的隊列 。

生產者

生產者鏈接到RabbitMQ，發送一條消息，而後退出。

# send.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();

// 定義一個名爲 direct_logs 的 direct 交換機
$channel->exchange_declare('direct_logs', 'direct', false, false, false);

// 從參數中獲取交換機的路由鍵
$severity = isset($argv[1]) && !empty($argv[1]) ? $argv[1] : 'info';

$data = implode(' ', array_slice($argv, 2));
if (empty($data)) {
    $data = "Hello World!";
}

// 
$msg = new AMQPMessage($data);

// 經過名爲 direct_logs 的 direct 交換機發送消息到隊列 (消息內容, 交換機, 路由鍵);
$channel->basic_publish($msg, 'direct_logs', $severity);

echo ' [x] Sent ', $severity, ':', $data, "\n";

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

消費者

消費者監聽來自RabbitMQ的消息，一般須要一直保持運行狀態以監聽消息。

# receive.php

<?php
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;

// 建立鏈接
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');

// 建立通道
$channel = $connection->channel();

// 定義一個名爲 direct_logs 的 direct 交換機
$channel->exchange_declare('direct_logs', 'direct', false, false, false);

// 建立一個隨機命名的新隊列，第三個參數爲關閉隊列持久化，第四個參數爲當聲明它的鏈接關閉時隊列會被自動刪除
list($queue_name, ,) = $channel->queue_declare("", false, false, true, false);

// 定義全部的綁定鍵
$severities = array_slice($argv, 1);
if (empty($severities)) {
    file_put_contents('php://stderr', "Usage: $argv[0] [info] [warning] [error]\n");
    exit(1);
}

// 將隨機命名的隊列綁定到 direct 交換機，生產者向交換機發送消息將被放到綁定的隊列中
foreach ($severities as $severity) {
    $channel->queue_bind($queue_name, 'direct_logs', $severity);
}

echo " [*] Waiting for logs. To exit press CTRL+C\n";

// 定義回調函數
$callback = function ($msg) {
    echo ' [x] ', $msg->delivery_info['routing_key'], ':', $msg->body, "\n";
};

// 第四個參數設爲true開啓自動消息確認，即投遞消息後馬上標記爲刪除
$channel->basic_consume($queue_name, '', false, true, false, false, $callback);

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

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

運行

打開一個終端，運行消費者，接收警告和錯誤的消息：

php receive.php warning error > logs_from_rabbit.log

打開另外一個終端，運行消費者，接收信息、警告和錯誤的消息：

php receive.php info warning error
# => [*] Waiting for logs. To exit press CTRL+C

打開另外一個終端，運行生產者，發送錯誤消息：

php send.php error "Run. Run. Or it will explode."

查看全部交換機

sudo rabbitmqctl list_exchanges

查看全部的綁定關係

sudo rabbitmqctl list_bindings