rabbitMQ发布/订阅模式、交换机之fanout类型

发布/订阅

我们将会做一些完全不同的事情——我们将会发送一条消息去多个Consumers（消费者）上。这种模式被称为“发布/订阅” 。

交换

生产者发送消息到交换机，交换机接收消息，它把这些消息推进Queue（队列） 交换类型：direct，topic，headers和fanout

创建一个这种类型的交换机并称他为logs

$channel->exchange_declare('logs', 'fanout', false, false, false);

fanout交换机：将它获取到的所有消息广播到它所知道的所有Queue（队列）中。

默认的交换机

空字符串（""）识别的默认的交换机

$channel->basic_publish($msg, '', 'hello');

使用了默认的，或者说是无名的交换机：消息会被路由到routing_key指定的Queue（队列），如果这个队列存在的化。routing_key就是basic_publish 的第三个参数。

发送消息到一个被命名（logs）的交换机上：

$channel->exchange_declare('logs', 'fanout', false, false, false);
$channel->basic_publish($msg, 'logs');

临时队列

当我们传给Queue（队列）名称参数一个空字符串时候，我们能创建一个非持久化队列(non-durable Queue)，并反回了一个（自动）生成的队列名称。

list($queue_name,,) = $channel->queue_declare('');

$queue_name变量包含了一个由RabbitMQ生成的随机名称，例如：amq.gen-JzTY20BRgKO-HjmUJj0wLg

一旦我们的Consumer（消费者）断开了链接，对应的Queue（队列）应该被自动删除。

绑定

交换机与队列之间的关系成为绑定（binding）

//logs交换机将会追加消息去我们的队列
$channel->queue_bind($queue_name, 'logs'); 

1、（emit_log.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
$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); 

// 发送消息到交换机上（而不是指定队列）
$channel->basic_publish($msg, 'logs'); 

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

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

2、（reveive_logs.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('logs', 'fanout', false, false, false); 

// 创建一个非持久化的队列并获取自动生成的队列名称
list($queue_name, ,) = $channel->queue_declare("", false, false, true, false); 

// 绑定队列到交换机
$channel->queue_bind($queue_name, 'logs'); 
echo " [*] Waiting for logs. To exit press CTRL+C\n"; 

$callback = function ($msg) { 
echo ' [x] ', $msg->body, "\n"; 
}; 

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

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

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