七、通过主题进行消息分发

<?php

/**

 * 发送消息

 */

​

$exchangeName = 'topic_logs';

$topic = empty($argv[1]) ? 'anonymous.info' : $argv[1]; // 主题

$message = empty($argv[2]) ? 'Hello World!' : $argv[2];

​

// 建立TCP连接

$connection = new AMQPConnection([

    'host' => 'localhost',

    'port' => '5672',

    'vhost' => '/',

    'login' => 'guest',

    'password' => 'guest'

]);

$connection->connect() or die("Cannot connect to broker!\n");

​

$channel = new AMQPChannel($connection);

$exchange = new AMQPExchange($channel);

$exchange->setName($exchangeName);

$exchange->setType(AMQP_EX_TYPE_TOPIC);

$exchange->declareExchange();

​

$exchange->publish($message, $topic);

echo "Message is sent: " . $message . "\n";

$connection->disconnect();

<?php

/**

 * 接收消息

 */

​

$exchangeName = 'topic_logs';

$topic = $argv[1];

​

// 建立TCP连接

$connection = new AMQPConnection([

    'host' => 'localhost',

    'port' => '5672',

    'vhost' => '/',

    'login' => 'guest',

    'password' => 'guest'

]);

$connection->connect() or die("Cannot connect to broker!\n");

​

$channel = new AMQPChannel($connection);

​

$exchange = new AMQPExchange($channel);

$exchange->setName($exchangeName);

$exchange->setType(AMQP_EX_TYPE_TOPIC);

$exchange->declareExchange();

​

$queue = new AMQPQueue($channel);

$queue->setFlags(AMQP_EXCLUSIVE);

$queue->declareQueue();

$queue->bind($exchangeName, $topic);

​

echo "Waiting for logs...\n";

while (TRUE) {

    $queue->consume('processLogs');

}

​

$connection->disconnect();

​

function processLogs($envelope, $queue) {

    $logs = $envelope->getBody();

    var_dump("Received: " . $logs);

    $queue->ack($envelope->getDeliveryTag()); // 手动发送ACK应答

}