6

以前我正在阅读队列中存在的所有消息，但现在我必须根据用户的选择（计数）返回特定数量的消息。

我尝试相应地更改 for 循环，但由于自动确认，它会读取所有消息。所以我尝试在配置文件中将其更改为手动。

在我的程序中，如何在阅读 msg 后手动确认消息（目前我正在使用 AmqpTemplate 接收并且我没有频道参考）？

    Properties properties = admin.getQueueProperties("queue_name");
    if(null != properties)
    {
        Integer messageCount = Integer.parseInt(properties.get("QUEUE_MESSAGE_COUNT").toString());          
        while(messageCount > 0)
        {
            Message msg = amqpTemplate.receive(queue_name);
            String value = new String(msg.getBody());
            
            valueList.add(value);
            messageCount--;
        }
}

任何帮助都是非常可观的，在此先感谢。

4

1 回答 1

12

您不能使用该receive()方法手动确认 - 将 aSimpleMessageListenerContainer用于具有 MANUAL ack 和 a 的事件驱动消费者ChannelAwareMessageListener。或者，使用模板的execute()方法让您可以访问Channel- 但是您将使用较低级别的 RabbitMQ API，而不是Message抽象。

编辑：

您需要学习底层的 RabbitMQ Java API 才能使用执行，但是这样的东西会起作用......

    final int messageCount = 3;
    boolean result = template.execute(new ChannelCallback<Boolean>() {

        @Override
        public Boolean doInRabbit(final Channel channel) throws Exception {
            int n = messageCount;
            channel.basicQos(messageCount); // prefetch
            long deliveryTag = 0;
            while (n > 0) {
                GetResponse result = channel.basicGet("si.test.queue", false);
                if (result != null) {
                    System.out.println(new String(result.getBody()));
                    deliveryTag = result.getEnvelope().getDeliveryTag();
                    n--;
                }
                else {
                    Thread.sleep(1000);
                }
            }
            if (deliveryTag > 0) {
                channel.basicAck(deliveryTag, true);
            }
            return true;
        }
    });
于 2015-04-15T14:04:30.140 回答