Search code examples
javaspring-bootrabbitmqamqpspring-amqp

How can make the RabbitMQ PrefetchCount work?

This is my RabbitListenerFactory:

  @Bean("workListenerFactory")
  public RabbitListenerContainerFactory myFactory(ConnectionFactory connectionFactory) {
      SimpleRabbitListenerContainerFactory containerFactory = new SimpleRabbitListenerContainerFactory();
      containerFactory.setConnectionFactory(connectionFactory);
      containerFactory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
      containerFactory.setDefaultRequeueRejected(true);
      containerFactory.setPrefetchCount(2);
  }

And this is my Consumer:

  @RabbitListener(queues = "test-todo-delete", containerFactory = "workListenerFactory")
  public void onAuditResult(Message message, Channel channel) throws IOException {
    String payload = new String(message.getBody(), StandardCharsets.UTF_8);
    log.info("get the message:{}  TimeStamp:{}", payload, new Date());
    channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
  }

In the RabbitMQ client, it looks that the configuration is OK (i sent the message in console): enter image description here

But my actual log:

2022-01-06 16:16:25.051 [ INFO] [cTaskExecutor-1] [TID: N/A] [c.p.o.usersignon.server.mq.TestConsumer ] : get the message:{"A":"1"}  TimeStamp:Thu Jan 06 16:16:25 CST 2022
2022-01-06 16:16:25.202 [ INFO] [cTaskExecutor-1] [TID: N/A] [c.p.o.usersignon.server.mq.TestConsumer ] : get the message:{"A":"1"}  TimeStamp:Thu Jan 06 16:16:25 CST 2022
2022-01-06 16:16:25.353 [ INFO] [cTaskExecutor-1] [TID: N/A] [c.p.o.usersignon.server.mq.TestConsumer ] : get the message:{"A":"1"}  TimeStamp:Thu Jan 06 16:16:25 CST 2022
2022-01-06 16:16:25.554 [ INFO] [cTaskExecutor-1] [TID: N/A] [c.p.o.usersignon.server.mq.TestConsumer ] : get the message:{"A":"1"}  TimeStamp:Thu Jan 06 16:16:25 CST 2022
2022-01-06 16:16:25.746 [ INFO] [cTaskExecutor-1] [TID: N/A] [c.p.o.usersignon.server.mq.TestConsumer ] : get the message:{"A":"1"}  TimeStamp:Thu Jan 06 16:16:25 CST 2022
2022-01-06 16:16:25.936 [ INFO] [cTaskExecutor-1] [TID: N/A] [c.p.o.usersignon.server.mq.TestConsumer ] : get the message:{"A":"1"}  TimeStamp:Thu Jan 06 16:16:25 CST 2022
2022-01-06 16:16:26.159 [ INFO] [cTaskExecutor-1] [TID: N/A] [c.p.o.usersignon.server.mq.TestConsumer ] : get the message:{"A":"1"}  TimeStamp:Thu Jan 06 16:16:26 CST 2022
2022-01-06 16:16:26.340 [ INFO] [cTaskExecutor-1] [TID: N/A] [c.p.o.usersignon.server.mq.TestConsumer ] : get the message:{"A":"1"}  TimeStamp:Thu Jan 06 16:16:26 CST 2022
2022-01-06 16:16:26.535 [ INFO] [cTaskExecutor-1] [TID: N/A] [c.p.o.usersignon.server.mq.TestConsumer ] : get the message:{"A":"1"}  TimeStamp:Thu Jan 06 16:16:26 CST 2022
2022-01-06 16:16:26.716 [ INFO] [cTaskExecutor-1] [TID: N/A] [c.p.o.usersignon.server.mq.TestConsumer ] : get the message:{"A":"1"}  TimeStamp:Thu Jan 06 16:16:26 CST 2022
2022-01-06 16:16:26.900 [ INFO] [cTaskExecutor-1] [TID: N/A] [c.p.o.usersignon.server.mq.TestConsumer ] : get the message:{"A":"1"}  TimeStamp:Thu Jan 06 16:16:26 CST 2022
2022-01-06 16:16:27.082 [ INFO] [cTaskExecutor-1] [TID: N/A] [c.p.o.usersignon.server.mq.TestConsumer ] : get the message:{"A":"1"}  TimeStamp:Thu Jan 06 16:16:27 CST 2022

Why is it not output twice per second?

Solution

  • Prefetch simply means how many un-acknowledged messages can be pending at the client at any time; as soon as you ack a message, the next one is sent.