2

我有一个生产者向主题交换发送消息。每条消息都包含一个路由键。（为原始图道歉）

       P
       |
       X
     /| |\
    / | | \
   /  | |  \
   Q1 Q2 Q3 Q4
   |   / /  /
   |  / /  /
   | / /  /
   |/ /  /
      C

我正在使用php-amqplib并尝试使用多个队列。我想要实现的是按顺序测试每个队列，看看它是否有消息，如果有，处理它，否则，继续下一个队列。此外，如果找到消息，请从 Q1 重新开始检查过程。以下代码不起作用，但将演示我想要做的逻辑。

$connection = new AMQPConnection(HOST, PORT, USER, PASS, VHOST);
$channel = $connection->channel();

$channel->exchange_declare('myexchange', 'topic', false, false, false);

$channel->queue_declare("Q1", false, true, false, false);
$channel->queue_bind("Q1", 'myexchange', 'priority.1');

$channel->queue_declare("Q2", false, true, false, false);
$channel->queue_bind("Q2", 'myexchange', 'priority.2');

$channel->queue_declare("Q3", false, true, false, false);
$channel->queue_bind("Q3", 'myexchange', 'priority.3');

$channel->queue_declare("DFQ4", false, true, false, false);
$channel->queue_bind("DFQ4", 'myexchange', 'priority.4');

$queues = array('Q1','Q2','Q3','Q4');

$priority = 0;
while (1) {

    $priority = ($priority<4)? $priority+1 : 0;

    $msg = $channel->basic_consume($queues[$priority], $consumer_tag, false, false, false, false);
    if(isset($msg->body)) {
        echo ' [x] ',$msg->delivery_info['routing_key'], "\n";
        $channel->basic_ack($msg->delivery_info['delivery_tag']);
        $priority = 0;
    }
}

$channel->close();
$connection->close();
4

1 回答 1

0

我对我的 .NET 项目有同样的要求。基本上，我有类似的方法，但我使用标头交换将发布的消息提供给这些队列。我创建的库订阅所有这些队列（具有相同的预取大小值）并将所有这些消息放入内部内存队列，这些消息可以按其优先级排序。所以在管道的末端，处理消息的实际代码将从内部队列中读取它们，具有较高优先级的消息将被处理，然后在较低优先级的消息之前被确认。

于 2014-01-25T09:50:49.877 回答