Сообщения в ручном режиме в RabbitMQ

Question

Раньше я читал все сообщения, присутствующие в очереди, но теперь мне нужно вернуть определенное количество сообщений на основе выбора (количества) пользователей.

Я пытаюсь изменить цикл for соответственно, но его чтение всего сообщения из-за автоматического подтверждения. Поэтому я попытался изменить его на ручной в файле конфигурации.

В моей программе как ack сообщение вручную после прочтения 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--; 
     } 
} 

Любая помощь очень заметна, спасибо заранее.

Answer 1

Вы не можете вручную оценить метод - используйте SimpleMessageListenerContainer для потребителя, управляемого событиями, с РУКОВОДСТВАМИ и ChannelAwareMessageListener. Или используйте метод шаблона execute(), который дает вам доступ к Channel - но тогда вы будете использовать API-интерфейс более низкого уровня RabbitMQ, а не абстракцию Message.

EDIT:

Вы должны изучить лежащую в основе 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; 
     } 
    });