RabbitMQ Fanout Exchange：实现消息的全量广播

<?php
require_once __DIR__ . '/vendor/autoload.php';

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

// 连接配置（请替换为你的实际信息）
$dbName = 'sanqing';
$dbPwd = '111111';
$tableName = 'order';

$connection = new AMQPStreamConnection('localhost', 5672, $dbName, $dbPwd, $tableName);
$channel = $connection->channel();

// 声明一个Fanout类型的交换机，名为 'exch'
$exc_name = 'exch';
$channel->exchange_declare($exc_name, 'fanout', false, false, false);

// 发送消息到交换机
$data = 'this is fanout message';
$msg = new AMQPMessage($data, ['delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT]);
$channel->basic_publish($msg, $exc_name);

// 注意：这里没有指定队列，因为消息是发给交换机的

echo " [x] 消息已发送：", $data, "\n";

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

<?php
require_once __DIR__ . '/vendor/autoload.php';

use PhpAmqpLib\Connection\AMQPStreamConnection;

// 连接配置（请与生产者保持一致）
$dbName = 'sanqing';
$dbPwd = '111111';
$tableName = 'order';

$connection = new AMQPStreamConnection('localhost', 5672, $dbName, $dbPwd, $tableName);
$channel = $connection->channel();

// 声明要绑定的Fanout交换机
$exc_name = 'exch';
$channel->exchange_declare($exc_name, 'fanout', false, false, false);

// 创建一个临时队列（名称为空，RabbitMQ会自动生成一个随机名称）
// 此队列会在消费者断开连接时自动删除
list($queue_name, ,) = $channel->queue_declare('', false, false, true, false);

// 将队列绑定到交换机
$channel->queue_bind($queue_name, $exc_name);

echo " [*] 等待消息。要退出请按 CTRL+C\n";

// 定义消息处理回调
$callback = function ($msg) {
    echo ' [x] 收到消息：', $msg->body, "\n";
    $msg->ack(); // 确认消息已处理
};

// 设置同时只处理一条消息
$channel->basic_qos(null, 1, null);
$channel->basic_consume($queue_name, '', false, false, false, false, $callback);

// 循环等待消息
while ($channel->is_open()) {
    $channel->wait();
}

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