生产者:
<?php
require_once dirname(__DIR__) . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
//创建RabbitMq连接
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
//创建信道
$channel = $connection->channel();
//创建叫logs的交换器
//$exchange=》交换器名称,$type=》交换器类型,$passive=》,$durable=》是否持久化,$auto_delete=》是否自动删除
$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();
消费者:
<?php
require_once dirname(__DIR__) . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
//创建RabbitMq连接
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
//创建信道
$channel = $connection->channel();
//创建叫logs的交换器
//$exchange=》交换器名称,$type=》交换器类型,$passive=》,$durable=》是否持久化,$auto_delete=》是否自动删除
$channel->exchange_declare('logs', 'fanout', false, false, false);
//创建空名称的队列,并获取队列名称
//$queue=》队列名称,$passive=》,$durable=》是否持久化,$exclusive=》是否排他,$auto_delete=》是否自动删除
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 (count($channel->callbacks)) {
$channel->wait();
}
//关闭信道
$channel->close();
//关闭连接
$connection->close();