RabbitMQ 按优先级从队列中消费消息

Question

我的想法是我想接收诸如“链接123456”、“代码1234”之类的消息。

链接号必须尽可能快地被消耗，而代码可以等待所有当前链接号被消耗。来自此处的RabbitMQ文档：https://www.rabbitmq.com/priority.html非常简短和干净，但对我一点帮助都没有。 当我遵循它时，消息完全忽略优先级并按发送顺序显示。

我尝试用Java实现它，如示例所示，但结果是相同的。

发件人代码：

优先级为 5 的发件人

<?php
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;

$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$table = [
    'maxPriority' => ['I', 10],
];
$channel->queue_declare('queue', false, true, false, false, false, $table);

$msg = new AMQPMessage('link: '.$_SERVER['argv'][1], [
                                              'delivery_mode' => 2,
                                              'priority' => 5,
                                            ]);
$channel->basic_publish($msg, '', 'queue');
$channel->close();
$connection->close();
?>

优先级为 1 的发件人

<?php
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;

$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$table = [
    'maxPriority' => ['I', 10],
];
$channel->queue_declare('queue', false, true, false, false, false, $table);

$msg = new AMQPMessage('code: '.$_SERVER['argv'][1], [
                                              'delivery_mode' => 2,
                                              'priority' => 1,
                                            ]);
$channel->basic_publish($msg, '', 'queue');
$channel->close();
$connection->close();
?>

接收者代码：

<?php
    require_once __DIR__ . '/vendor/autoload.php';
    use PhpAmqpLib\Connection\AMQPStreamConnection;
    use PhpAmqpLib\Message\AMQPMessage;
    use PhpAmqpLib\Wire\AMQPTable;
    function s($str, $charset = null) {
        return new \Delight\Str\Str($str, $charset);
    }
    $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
    $channel = $connection->channel();
    $table = [
        'maxPriority' => ['I', 10],
    ];
    $channel->queue_declare('queue', false, true, false, false, false, $table);
    echo ' [*] Waiting for messages. To exit press CTRL+C', "\n";
    $callback = function($msg){
        file_put_contents('rabbit_logs.txt', $msg->body."\r\n", FILE_APPEND);
        if( (new \Delight\Str\Str($msg->body))->startsWith('code: ')==1){
            echo " [x] Received ", $msg->body, "\n";
            sleep(1);
            echo " [x] Code sent", "\n";
        }
        else{
            if( (new \Delight\Str\Str($msg->body))->startsWith('lead: ')==1){
                echo " [x] Received ", $msg->body, "\n";
                sleep(1);
                echo " [x] lead sent", "\n";
            }
            else{
                if( (new \Delight\Str\Str($msg->body))->startsWith('link: ')==1){
                    echo " [x] Received ", $msg->body, "\n";
                    sleep(1);
                    echo " [x] link sent", "\n";
                }
                else{
                    echo " [x] Received ", $msg->body, "\n";      
                    echo " [x] Nothing sent", "\n";
                }
            }
        }
        $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
    };
    $channel->basic_qos(null, 1, null);
    $channel->basic_consume('queue', '', false, false, false, false, $callback);
    while(count($channel->callbacks)) {
        $channel->wait();
    }
    $channel->close();
    $connection->close();
?>

Answer 1

您需要使用

x-max-priority

参数并构建参数表，如 此演示中所示。