五、消息订阅（Publish/Subscribe）

之前都是将消息发送到同一个 Consumer，而现在我们需将其发送到多个 Consumer。

我们将创建一个日志系统，它包含两个部分：第一个部分负责发出log（Producer），第二个部分负责接收并打印（Consumer）。我们将构建两个 Consumer，第一个将 log 写到物理磁盘上；第二个将 log 输出到屏幕。

"Fanout" not telling an exchange to distribute messages to different consumers, but to different queues. So, you need at least two queues binding to a "fanout" exchange. Then let your two consumers get message from those two queues, one consumer to one queue.

Exchange

Exchange 决定将 Message 发送到具体的 Queue，至于是发送给一个 Queue 还是多个 Queue，则需要通过 Exchange 的类型类决定。Exchange 分为三种类型：direct、topic 和 fanout。fanout 就是广播模式，会将 Message 都放到它所知道的所有 Queue 中：

$exchange->setType(AMQP_EX_TYPE_FANOUT);

现在我们可以直接通过 Exchange，而不需要 routing_key 来发送 Message 了：

$exchange->publish($message);

临时队列

截至现在，我们用的 Queue 都是有名字的。使用有名字的 Queue，使得在 Producer 和 Consumer 之前共享 Queue 成为可能。

在我们的日志系统中，不需要有名字的队列，要实现这个目标，需要在声明队列时不指定名称，而由系统随机分配：

$queue = new AMQPQueue($channel);
//$queue->setName($queueName);
$queue->setFlags(AMQP_EXCLUSIVE);
$queue->declareQueue();
//$queue->bind($exchangeName, $routeKey);

这时，通过 $queue->getName() 获取到的队列名称是随机生成的。

绑定

建立 Exchange 与 Queue 之间的绑定：

$queue->bind($exchangeName);

演示代码

emit_logs.php

<?php
/**
 * 发送消息
 */

$exchangeName = 'logs';
$message = 'Hello World!';

// 建立TCP连接
$connection = new AMQPConnection([
    'host' => 'localhost',
    'port' => '5672',
    'vhost' => '/',
    'login' => 'guest',
    'password' => 'guest'
]);
$connection->connect() or die("Cannot connect to the broker!\n");

try {
    $channel = new AMQPChannel($connection);

    $exchange = new AMQPExchange($channel);
    $exchange->setName($exchangeName);
    $exchange->setType(AMQP_EX_TYPE_FANOUT);
    $exchange->declareExchange();

    echo 'Send Message: ' . $exchange->publish($message) . "\n";
    echo "Message Is Sent: " . $message . "\n";
} catch (AMQPConnectionException $e) {
    var_dump($e);
}

// 断开连接
$connection->disconnect();

receive_logs.php

<?php
/**
 * 接收消息
 */

$exchangeName = 'logs';

// 建立TCP连接
$connection = new AMQPConnection([
    'host' => 'localhost',
    'port' => '5672',
    'vhost' => '/',
    'login' => 'guest',
    'password' => 'guest'
]);
$connection->connect() or die("Cannot connect to the broker!\n");

$channel = new AMQPChannel($connection);

$exchange = new AMQPExchange($channel);
$exchange->setName($exchangeName);
$exchange->setType(AMQP_EX_TYPE_FANOUT);
$exchange->declareExchange();

$queue = new AMQPQueue($channel);
$queue->setFlags(AMQP_EXCLUSIVE);
$queue->declareQueue();
$queue->bind($exchangeName);

var_dump("Waiting for message...");

// 消费队列消息
while(TRUE) {
    $queue->consume('processMessage');
}

// 断开连接
$connection->disconnect();

function processMessage($envelope, $queue) {
    $msg = $envelope->getBody();
    var_dump("Received: " . $msg);
    $queue->ack($envelope->getDeliveryTag()); // 手动发送ACK应答
}

演示流程

打开两个终端，一个消费者队列负责将日志写入文件：

php receive_logs.php > logs_from_rabbit.log 

一个负责将日志输出到屏幕：

php receive_logs.php

然后再打开一个终端，将日志信息发送到所有队列：

php emit_logs.php

这样，会发现所有队列会同时接收到日志并进行相应的处理。