釋出/訂閱，使用扇型交換機（fanout）

composer.json

### composer.json
 {
    "require": {
        "php-amqplib/php-amqplib": ">=2.9.0"
    }
}

釋出端（Publish）

/**
 * rabbitmq
 * 釋出／訂閱
 * Publish
 * https://github.com/rabbitmq/rabbitmq-tutorials
 * https://www.rabbitmq.com/tutorials/tutorial-three-php.html
 */

defined('DS') or define('DS', DIRECTORY_SEPARATOR);
require_once __DIR__. DS . 'vendor' .DS.'autoload.php';

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

$connection = new AMQPStreamConnection('192.168.0.83', 5672, 'admin', 'admin');
$channel = $connection->channel();

// 建立一個fanout型別的交換機，命名為logs
// 扇型交換機（fanout），它把訊息傳送給它所知道的所有佇列
$channel->exchange_declare('logs', 'fanout', false, false, false);

$data = implode('...', array_slice($argv, 1));
if (empty($data)) {
    $data = 'hello publish,subscribe!';
}

$msg = new AMQPMessage($data);
$channel->basic_publish($msg, 'logs');

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

訂閱端（Subscribe）

/**
 * rabbitmq
 * 釋出／訂閱
 * Subscribe
 * https://github.com/rabbitmq/rabbitmq-tutorials
 * https://www.rabbitmq.com/tutorials/tutorial-three-php.html
 */

defined('DS') or define('DS', DIRECTORY_SEPARATOR);
require_once __DIR__. DS . 'vendor' .DS.'autoload.php';

use PhpAmqpLib\Connection\AMQPStreamConnection;

$connection = new AMQPStreamConnection('192.168.0.83', 5672, 'admin', 'admin');
$channel = $connection->channel();

// 建立一個fanout型別的交換機，命名為logs
// 扇型交換機（fanout），它把訊息傳送給它所知道的所有佇列
$channel->exchange_declare('logs', 'fanout', false, false, false);

// 定義臨時佇列
// 當與消費者（consumer）斷開連線的時候，這個佇列被立即刪除
list($queue_name, ,) = $channel->queue_declare('', false, false, true, false);

// logs交換機將會把訊息新增到我們的佇列中
$channel->queue_bind($queue_name, 'logs');

// 訂閱回撥函式
$callback = function($msg){
    echo 'Subscribe:', $msg->body, PHP_EOL;
};

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

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

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