<?php

/**
 * RPC客户端
 */
$routing_key = 'rpc_queue';
$num = empty($argv[1]) ? 0 : intval($argv[1]);

// 建立TCP连接
$connection = new AMQPConnection([
    'host' => 'localhost',
    'port' => '5672',
    'vhost' => '/',
    'login' => 'guest',
    'password' => 'guest'
]);
$connection->connect() or die("Cannot connect to the broker!\n");

$channel = new AMQPChannel($connection);

$client_queue = new AMQPQueue($channel);
$client_queue->setFlags(AMQP_EXCLUSIVE);
$client_queue->declareQueue();
$callback_queue_name = $client_queue->getName();

$corr_id = uniqid();
$properties = [
    'correlation_id' => $corr_id,
    'reply_to' => $callback_queue_name
];

$exchange = new AMQPExchange($channel);
$exchange->publish($num, $routing_key, AMQP_NOPARAM, $properties);

$client_queue->consume(function($envelope, $queue) use ($corr_id){
    if ($envelope->getCorrelationId() == $corr_id) {
        $msg = $envelope->getBody();
        var_dump('Received Data: ' . $msg);
        $queue->nack($envelope->getDeliveryTag());
        return false;
    }
});

// 断开连接
$connection->disconnect();

<?php

/**
 * RPC服务端
 */
$routing_key = 'rpc_queue';

// 建立TCP连接
$connection = new AMQPConnection([
    'host' => 'localhost',
    'port' => '5672',
    'vhost' => '/',
    'login' => 'guest',
    'password' => 'guest'
]);
$connection->connect() or die("Cannot connect to the broker!\n");

$channel = new AMQPChannel($connection);
$channel->setPrefetchCount(1);

$server_queue = new AMQPQueue($channel);
$server_queue->setName($routing_key);
$server_queue->declareQueue();

$exchange = new AMQPExchange($channel);

$server_queue->consume(function($envelope, $queue) use ($exchange){
    $num = intval($envelope->getBody());
    $response = fib($num);
    $exchange->publish($response, $envelope->getReplyTo(), AMQP_NOPARAM, [
        'correlation_id' => $envelope->getCorrelationId(),
    ]);
    $queue->ack($envelope->getDeliveryTag());
});

// 断开连接
$connection->disconnect();

// 斐波那契函数
function fib($num) {
    if ($num == 0)
        return 1;
    else if ($num == 1)
        return 1;
    else
        return fib($num - 1) + fib($num - 2);
}

php rpc_server.php

php rpc_client.php 1
php rpc_client.php 5
php rpc_client.php 15
php rpc_client.php 20