共计 2247 个字符,预计需要花费 6 分钟才能阅读完成。
公布订阅的利用场景通常为一个生产者、一个交换机,多个消费者各自创立队列绑定到交换机上订阅音讯并生产。
公布订阅通常配合以下设置独特应用:
- 消费者创立的队列随机命名即可
- 敞开队列长久化
- 敞开音讯长久化
- 消费者创立的队列在连贯敞开时主动删除
- 开启主动音讯确认
装置依赖
# composer.json
{
"require": {"php-amqplib/php-amqplib": ">=3.0"}
}
> composer.phar install
模式构造
在 RabbitMQ 消息传递模型中,生产者是不会向队列间接发送音讯的,只能将音讯发送给交换机。
交换机接管来自生产者的音讯,而后将它们推送到队列中。
公布订阅应用的是 fanout
交换机,这个交换机非常简单,将它收到的所有音讯播送到它绑定的所有队列。
生产者
生产者连贯到 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();
// 定义一个名为 logs 的 fanout 播送交换机
$channel->exchange_declare('logs', 'fanout', false, false, false);
$data = implode(' ', array_slice($argv, 1));
if (empty($data)) {$data = "info: Hello World!";}
//
$msg = new AMQPMessage($data);
// 将音讯发送到名为 logs 的 fanout 播送交换机 (音讯内容, 交换机, 路由键);
$channel->basic_publish($msg, 'logs');
echo '[x] Sent', $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();
// 定义一个名为 logs 的 fanout 播送交换机
$channel->exchange_declare('logs', 'fanout', false, false, false);
// 创立一个随机命名的新队列,第三个参数为敞开队列长久化,第四个参数为当申明它的连贯敞开时队列会被主动删除
list($queue_name, ,) = $channel->queue_declare("", false, false, true, false);
// 将随机命名的队列绑定到 fanout 播送交换机,生产者向交换机发送音讯将被播送到绑定的队列中
$channel->queue_bind($queue_name, 'logs');
echo "[*] Waiting for messages. To exit press CTRL+C\n";
// 定义回调函数
$callback = function ($msg) {echo '[x]', $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 > logs_from_rabbit.log
关上另一个终端,运行消费者,将日志输入到终端:
php receive.php
# => [*] Waiting for messages. To exit press CTRL+C
# => [x] Received 'Second message..'
# => [x] Received 'Fourth message....'
关上另一个终端,运行生产者:
php send.php
查看所有交换机
sudo rabbitmqctl list_exchanges
查看所有的绑定关系
sudo rabbitmqctl list_bindings
# => Listing bindings ...
# => logs exchange amq.gen-JzTY20BRgKO-HjmUJj0wLg queue []
# => logs exchange amq.gen-vso0PVvyiRIL2WoV3i48Yg queue []
# => ...done.
正文完