如何轮询RabbitMQ以连续获取优先级顺序的消息？

Question

我们可以使RabbitMQ分布式优先级队列通过安装插件Rabbitmq-优先级队列从https://www.rabbitmq.com/community-plugins.html.我将元素推送到队列中（每个元素都有一个优先级），并且我能够根据需要在消费者中接收队列的内容-较高优先级的元素首先出现。

问题是，当这种情况持续发生时，优先级轮询概念不起作用：

1. 运行发布服务器以填充队列中具有不同优先级的3个项目
2. 使用队列中的消息-工作正常-按优先级使用。现在消费者等待队列中的任何消息，此时队列为空
3. 我再次运行publisher来填充大约5个元素
4. 使用者不会优先使用队列中的5项，而是按照步骤3发布者发布的顺序使用

我需要的是，在队列的所有内容中具有最大优先级的队列项目的每次轮询都应该首先出现。

有人能告诉我这里发生了什么错误吗？谢谢

以下是出版商和消费者（Java）的片段：

出版商

public class RabbitMQPublisher {
    private static final String QUEUE = "my-priority-queue-3";
    public static void main(String[] argv) throws Exception {
        final ConnectionFactory factory = new ConnectionFactory();
        final Connection conn = factory.newConnection();
        final Channel ch = conn.createChannel();
        final Map<String, Object> args = new HashMap<String, Object>();
        args.put("x-max-priority", 100);
        ch.queueDeclare(QUEUE, true, false, false, args);
        publish(ch, 24);
        publish(ch, 11);
        publish(ch, 75);
        //second run
        //publish(ch, 27);
        //publish(ch, 77);
        //publish(ch, 12);
        conn.close();
    }

    private static void publish(Channel ch, int priority) throws IOException {
        final BasicProperties props = MessageProperties.PERSISTENT_BASIC.builder().priority(priority).build();
        final String body = "message with priority " + priority;
        ch.basicPublish("", QUEUE, props, body.getBytes());
    }

消费者

while (true) {
        final QueueingConsumer.Delivery delivery = consumer.nextDelivery();
        final String message = new String(delivery.getBody());
        System.out.println(message);
    }

输出：

message with priority 75
message with priority 24
message with priority 11
message with priority 27
message with priority 77
message with priority 12

Answer 1

我能够使用basicGet来轮询队列而不是消费者来解决这个问题。nextDelivery（）<代码>最终字符串消息=新字符串（channel.basicGet（QUEUE_NAME，true））。getBody（）） 这将从队列中提取具有最高优先级的项目。