RabbitMQ RPC：独占队列锁定@ PHP

Question

我正在尝试使用类似于此示例的 RabbitMQ 在 PHP 上构建 RPC 服务：http: //www.rabbitmq.com/tutorials/tutorial-six-java.html 我正在使用此 PECL 扩展：http://pecl。 php.net/package/amqp（版本 1.0.3）

问题是，当我向服务器添加标志 AMQP_EXCLUSIVE 时，我的回调队列（在客户端脚本中声明）被锁定。

这是我的服务器

// connect to server
$cnn = new AMQPConnection('...');
$cnn->connect();
$channel = new AMQPChannel($cnn);
// create exchange
$exchangeName = 'k-exchange';
$exchange = new AMQPExchange($channel);
$exchange->setName($exchangeName);
$exchange->setType(AMQP_EX_TYPE_DIRECT);
$exchange->declare();

// declare queue to consume messages from
$queue = new \AMQPQueue($channel);
$queue->setName('tempQueue');
$queue->declare();

// start consuming messages
$queue->consume(function($envelope, $queue)
    use ($channel, $exchange) {

    // create callback queue
    $callbackQueue = new \AMQPQueue($channel);
    $callbackQueue->setName($envelope->getReplyTo());
    $callbackQueue->setFlags(AMQP_EXCLUSIVE); // set EXCLUSIVE flag

    /* WARNING: Following code line causes error. See rabbit logs below:
     *  connection <0.1224.10>, channel 1 - error:
     *  {amqp_error,resource_locked,
     *  "cannot obtain exclusive access to locked queue 'amq.gen-Q6J...' in vhost '/'",
     *  'queue.bind'}
     */
    $callbackQueue->bind($exchange->getName(), 'rpc_reply');

    // trying to publish response back to client's callback queue
    $exchange->publish(
        json_encode(array('processed by remote service!')),
        'rpc_reply',
        AMQP_MANDATORY & AMQP_IMMEDIATE
    );

    $queue->ack($envelope->getDeliveryTag());
});

这是我的 Client.php

// connect to server
$cnn = new AMQPConnection('...');
$cnn->connect();
$channel = new AMQPChannel($cnn);
// create exchange
$exchangeName = 'k-exchange';
$exchange = new AMQPExchange($channel);
$exchange->setName($exchangeName);
$exchange->setType(AMQP_EX_TYPE_DIRECT);
$exchange->declare();

// create a queue which we send messages to server via
$queue = new \AMQPQueue($channel);
$queue->setName('tempQueue');
$queue->declare();

// binding exchange to queue
$queue->bind($exchangeName, 'temp_action');

// create correlation_id
$correlationId = sha1(time() . rand(0, 1000000));

// create anonymous callback queue to get server response response via
$callbackQueue = new \AMQPQueue($channel);
$callbackQueue->setFlags(AMQP_EXCLUSIVE); // set EXCLUSIVE flag
$callbackQueue->declare();

// publishing message to exchange (passing it to server)
$exchange->publish(
    json_encode(array('process me!')),
    'temp_action',
    AMQP_MANDATORY,
    array(
        'reply_to' => $callbackQueue->getName(), // pass callback queue name
        'correlation_id' => $correlationId
    )
);

// going to wait for remote service complete tasks. tick once a second
$attempts = 0;
while ($attempts < 5)
{
    echo 'Attempt ' . $attempts . PHP_EOL;
    $envelope = $callbackQueue->get();
    if ($envelope) {
        echo 'Got response! ';
        print_r($envelope->getBody());
        echo PHP_EOL;
        exit;
    }

    sleep(1);
    $attempts++;
}

所以最后我只在 RabbitMQ 的日志中看到错误：

connection <0.1224.10>, channel 1 - error:
{amqp_error,resource_locked,
    "cannot obtain exclusive access to locked queue 'amq.gen-Q6J...' in vhost '/'", 
    'queue.bind'}

问题：在 Server.php 中创建回调队列对象的正确方法是什么？看来我的 Server.php 与 RabbitMQ 服务器的连接与 Client.php 不同。我应该在这里做什么？我应该如何在 Server.php 端“共享”相同的（到 Client.php 的）连接。

更新 这里有更多 RabbitMQ 日志

我的 Server.php 连接（ID 为：<0.22322.27>）

=INFO REPORT==== 20-Jun-2012::13:30:22 ===
    accepting AMQP connection <0.22322.27> (127.0.0.1:58457 -> 127.0.0.1:5672)

我的 Client.php 连接（ID 为：<0.22465.27>）

=INFO REPORT==== 20-Jun-2012::13:30:38 ===
    accepting AMQP connection <0.22465.27> (127.0.0.1:58458 -> 127.0.0.1:5672)

现在我看到 Server.php 导致错误：

=ERROR REPORT==== 20-Jun-2012::13:30:38 ===
    connection <0.22322.27>, channel 1 - error:
{amqp_error,resource_locked,
"cannot obtain exclusive access to locked queue 'amq.gen-g6Q...' in vhost '/'",
'queue.bind'}

我的假设 我怀疑，由于 Client.php 和 Server.php 不共享具有相同 Id 的连接，因此它们不可能都使用 Client.php 中声明的独占队列

Answer 1

您的实施存在一些问题：

1. 兑换声明
2. 手动设置回复队列而不是使用临时队列
3. 双向使用 AMQP_EXCLUSIVE

兑换声明

您不需要声明交换器 (AMQPExchange) 来发布消息。在此 RPC 示例中，您需要使用它作为广播消息的方式（例如临时队列或临时交换）。所有通信都将直接发生在 QUEUE 上，理论上绕过交换。

$exchange = new AMQPExchange($channel);
$exchange->publish(...);

队列和回复：

当您将 AMQPQueue::setName() 与 AMQPQueue::declare() 一起使用时，您将绑定到具有用户定义名称的队列。如果声明队列时没有名称，则这称为临时队列。当您需要从特定路由键接收广播消息时，这非常有用。因此，RabbitMQ / AMQP 会生成一个随机临时名称。由于队列名称是为给定实例专门使用信息而创建的，因此，它会在连接关闭时被丢弃。

当 RPC 客户端想要发布消息 (AMQPExchange::publish()) 时，它必须指定回复作为发布参数之一。这样，RPC服务器在收到请求时就可以取到随机生成的名称。它使用回复名称作为服务器将回复给定客户端的队列的名称。除了临时队列名称之外，实例还必须发送correlationId，以确保它收到的回复消息对于请求实例是唯一的。

客户

$exchange = new AMQPExchange($channel);

$rpcServerQueueName = 'rpc_queue';

$client_queue = new AMQPQueue($this->channel);
$client_queue->setFlags(AMQP_EXCLUSIVE);
$client_queue->declareQueue();   
$callbackQueueName = $client_queue->getName(); //e.g. amq.gen-JzTY20BRgKO-HjmUJj0wLg

//Set Publish Attributes
$corrId = uniqid();
$attributes = array(
    'correlation_id' => $corrId,
    'reply_to'       => $this->callbackQueueName
);  

$exchange->publish(
    json_encode(['request message']),
    $rpcServerQueueName,
    AMQP_NOPARAM,
    $attributes
);  

//listen for response
$callback = function(AMQPEnvelope $message, AMQPQueue $q) {
    if($message->getCorrelationId() == $this->corrId) {
        $this->response = $message->getBody();
        $q->nack($message->getDeliveryTag());
        return false; //return false to signal to consume that you're done. other wise it continues to block
    }   
};  

$client_queue->consume($callback);

服务器

$exchange = new AMQPExchange($channel);

$rpcServerQueueName = 'rpc_queue';


$srvr_queue = new AMQPQueue($channel);
$srvr_queue->setName($rpcServerQueueName); //intentionally declares the rpc_server queue name
$srvr_queue->declareQueue();
...
$srvr_queue->consume(function(AMQPEnvelope $message, AMQPQueue $q) use (&$exchange) {

    //publish with the exchange instance to the reply to queue
    $exchange->publish(
        json_encode(['response message']),  //reponse message
        $message->getReplyTo(),             //get the reply to queue from the message
        AMQP_NOPARAM,                       //disable all other params
        $message->getCorrelationId()        //obtain and respond with correlation id
    );

    //acknowledge receipt of the message
    $q->ack($message->getDeliveryTag());
});

AMQP_EXCLUSIVE

在这种情况下，EXCLUSIVE 仅用于每个实例的 Rpc 客户端临时队列，以便它可以发布消息。换句话说，客户端为自己创建一个一次性临时队列，专门从 RPC 服务器接收应答。这确保没有其他通道线程可以在该队列上发布。它仅对客户端及其响应者锁定。需要注意的是，AQMP_EXCLUSIVE 不会阻止 RPC 服务器对客户端的回复队列进行响应。AMQP_EXCLUSIVE 属于尝试发布到同一队列资源的两个独立线程（通道实例）。当发生这种情况时，队列基本上被锁定以用于后续连接。交换声明也会发生相同的行为。

@Denis：在这种情况下，您的实现在某种程度上是正确的

不好 - 不要在服务器中重新声明队列。这就是客户的工作

$callbackQueue = new \AMQPQueue($channel);
$callbackQueue->setName($envelope->getReplyTo());
$callbackQueue->setFlags(AMQP_EXCLUSIVE); // set EXCLUSIVE flag 
...
$callbackQueue->bind($exchange->getName(), 'rpc_reply');

您正在尝试绑定到名为 tempQueue 的队列。但是您已经在 client.php 中创建了一个名为 tempQueue 的队列。根据首先启动的服务，另一个服务将抛出错误。因此，您可以删除所有内容并保留最后一部分：

// trying to publish response back to client's callback queue
$exchange->publish(
    json_encode(array('processed by remote service!')),
    'rpc_reply', //<--BAD Should be: $envelope->getReplyTo()
    AMQP_MANDATORY & AMQP_IMMEDIATE
);

然后修改上面的内容，替换为：

'rpc_reply'

 with

 $envelope->getReplyTo()

不要在客户端声明队列名称

// create a queue which we send messages to server via
$queue = new \AMQPQueue($channel);
//$queue->setName('tempQueue'); //remove this line
//add exclusivity
$queue->setFlags(AMQP_EXCLUSIVE);
$queue->declare();

//no need for binding... we're communicating on the queue directly
//there is no one listening to 'temp_action' so this implementation will send your message into limbo
//$queue->bind($exchangeName, 'temp_action'); //remove this line