rabbitMQ交换机之Direct类型

绑定键

绑定可以携带一个额外的bingding_key参数

$binding_key = 'black';
$channel->queue_bind($queue_name, $exchange_name, $binding_key);

注意：

1. routing_key就是basic_publish 的第三个参数是发送方（生产者）
2. bingding_key就是queue_bind的第三个参数是接收方（消费者）
3. 当Queue（队列）的binding key与消息的routing key完全匹配的时候，这条消息才会进入对应的路由

直接交换（Direct exchange）

上一个教程中的日志系统广播所有消息到所有的Consumer（消费者）。我们打算拓展它，让它可以根据消息的严重程度过滤消息。例如我们可能希望写日志的脚本仅把收到的严重错误的日志消息写到磁盘上，从而不把磁盘空间浪费在过多的警告或者信息类型的日志消息。

1. direct类型的交换机X和绑定在它上面的两个Queue（队列）
2. 第一个队列通过orange绑定键（binding key）来绑定
3. 第二个则拥有两个绑定，一个是通过black绑定键，另外一个是通过green绑定键

一条带有orange路由键（routing key）的消息发送到交换机会被路由到Queue（队列）Q1。带有black或者green路由键的消息将会被路由到Q2。所有其他的消息将会被丢弃。

多重绑定

在这一情况，direct路由将会像fanout路由那样广播消息到所有匹配的Queue（队列）上。一条拥有black路由键的消息将会被同时分发到Q1和Q2

发出日志

//创建一个direct类型的名为direct_logs的交换机
$channel->exchange_declare('direct_logs', 'direct', false, false, false);
//发送一条消息 
//“severity”可以是“info”，“warning”，“error”中的一个,表示日志发送的严重等级
$channel->basic_publish($msg, 'direct_logs', $severity);

订阅

//为我们感兴趣的每一个日志等级（severity）创建一个新的绑定
//当queue_bind的$severity匹配到basic_publish的severity相同等级的时候，才会收到对应的日志消息
foreach($severities as $severity) {
    $channel->queue_bind($queue_name, 'direct_logs', $severity);
}

如图：

<?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模式路由
$channel->exchange_declare('direct_logs', 'direct', false, false, false);

// 从命令行接收参数作为日志等级，默认为info
$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);

// 发送消息到对应的路由，并携带一个routing key
$channel->basic_publish($msg, 'direct_logs', $severity);

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

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

emit_log_direct.php源码

<?php

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

$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();

// 定义和生产者一样的direct交换机并命名为direct_logs
$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);
}

foreach ($severities as $severity) {
    // 绑定多个binding key
    $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";
};

$channel->basic_consume($queue_name, '', false, true, false, false, $callback);

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

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

receive_logs_direct.php源码