2015-07-11 247 views
2

我在我的日食中运行优先队列java program,我遇到了问题,第一次得到正确的答案。还有一次,我在队列中增加了一条消息,但这次我得到了不同的结果。RabbitMQ消息优先级队列不工作

public static void main(String[] argv) throws Exception { 
    ConnectionFactory factory = new ConnectionFactory(); 
     Connection conn = factory.newConnection(); 
     Channel ch = conn.createChannel(); 
     Map<String, Object> args = new HashMap<String, Object>(); 
     args.put("x-max-priority", 10); 
     ch.queueDeclare(QUEUE_UPDATE, true, false, false, args); 

     publish(ch, 141); 
     publish(ch, 250); 

     final CountDownLatch latch = new CountDownLatch(2); 
     ch.basicConsume(QUEUE_UPDATE, true, new DefaultConsumer(ch) { 
      public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { 
       System.out.println("Received " + new String(body)); 
       latch.countDown(); 
      } 
     }); 

     latch.await(); 
     conn.close();   
     System.out.println("Finished"); 
} 

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

正确的输出:

Received update-queue message with priority 250 
Received update-queue message with priority 141 
Finished 

增加了一个队列消息

  publish(ch, 141);  
     publish(ch, 250); 
     publish(ch, 110); // newly added 

预期输出

Received update-queue message with priority 250 
Received update-queue message with priority 141 
Received update-queue message with priority 110 
Finished 

实际输出

Received update-queue message with priority 141 
Received update-queue message with priority 250 
Received update-queue message with priority 110 
Finished 

这是怎么回事?我做错了什么?

+0

删除队列并尝试使用:'args.put(“x-max-priority”,250);' – Gabriele

+0

对于遇到此问题的C#开发人员,请参阅http://stackoverflow.com/questions/29221020/rabbitmq-3-5-and-message-priority/38999171#38999171。简而言之,“push api”不尊重.Priority。 – granadaCoder

回答

2

我遇到了同样的问题。对我而言有效的是定义一个由consumer prefetch定义的限制,例如channel.basicQos(1);。 如果您没有设置此限制,那么消息会在消费者到达代理时传递给消费者,因此它们不会使用优先级进行排序。

当您设置一个下限时,代理一次不会发送比此限制更多的消息,从而在交付之前对消息进行排序。

+0

非常感谢Toresan!如果我可以将它标记为正确答案,它就可以工作100%! –