locking - RabbitMQ 远程过程调用 : Exclusive queues locking @ 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() 一起使用时，您将绑定(bind)到一个具有用户定义名称的队列。如果您声明的队列没有名称，则称为临时队列。当您需要从特定路由键接收广播消息时，这很有用。为此，RabbitMQ/AMQP 生成一个随机的临时名称。由于队列名称是为给定实例专门使用信息而创建的，因此为了自身的利益，它会在连接关闭时被处理掉。

当 RPC 客户端想要发布消息 (AMQPExchange::publish()) 时，它必须将回复指定为发布参数之一。这样，RPC 服务器在收到请求时就可以获取随机生成的名称。它使用回复名称作为 QUEUE 的名称，服务器将在该 QUEUE 上回复给定的客户端。除了临时队列名称，实例还必须发送一个 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 服务器接收一个应答。这确保没有其他 channel 线程可以在该队列上发布。它仅对客户端及其响应者锁定。请务必注意，AQMP_EXCLUSIVE 不会阻止 RPC 服务器响应客户端的回复队列。 AMQP_EXCLUSIVE 与尝试发布到同一队列资源的两个独立线程( channel 实例)有关。发生这种情况时，队列实质上已为后续连接锁定。交换声明也会发生相同的行为。

@Denis:在这种情况下，您的实现在一定程度上是正确的

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

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

您正在尝试绑定(bind)到名为 tempQueue 的 QUEUE。但是您已经在 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