Spring AMQP-@RabbitListener轮询是否在兜帽下？

Question

我想异步处理来自AMQP/RabbitMQ队列的消息。我已经为此实现了一个RabbitListener方法（来自spring rabbit），但似乎这个侦听器实际上是在我的队列中进行轮询。这是意料之中的吗？我本以为监听器会以某种方式收到RabbitMQ的通知，而不必进行轮询。

如果这是意料之中的，我是否也可以在不进行轮询的情况下使用Spring AMQP异步消费消息？

当我发送消息时，听众会正确地接收到它。我仍然看到连续的日志消息流，表明侦听器继续轮询空队列：

…
15:41:10.543 [pool-1-thread-3] DEBUG o.s.a.r.l.BlockingQueueConsumer - ConsumeOK : Consumer: tags=[{amq.ctag-bUsK4KQN6_QHzf8DoDC_ww=myQueue}], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.1.1:5672/,1), acknowledgeMode=MANUAL local queue size=0
15:41:10.544 [main] DEBUG o.s.a.r.c.CachingConnectionFactory - Creating cached Rabbit Channel from AMQChannel(amqp://guest@127.0.1.1:5672/,2)
15:41:10.545 [main] DEBUG o.s.amqp.rabbit.core.RabbitTemplate - Executing callback on RabbitMQ Channel: Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.1.1:5672/,2)
15:41:10.545 [main] DEBUG o.s.amqp.rabbit.core.RabbitTemplate - Publishing message on exchange [], routingKey = [myQueue]
Sent: Hello World
15:41:10.559 [pool-1-thread-4] DEBUG o.s.a.r.l.BlockingQueueConsumer - Storing delivery for Consumer: tags=[{amq.ctag-bUsK4KQN6_QHzf8DoDC_ww=myQueue}], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.1.1:5672/,1), acknowledgeMode=MANUAL local queue size=0
15:41:10.560 [SimpleAsyncTaskExecutor-1] DEBUG o.s.a.r.l.BlockingQueueConsumer - Received message: (Body:'Hello World'MessageProperties [headers={}, timestamp=null, messageId=null, userId=null, appId=null, clusterId=null, type=null, correlationId=null, replyTo=null, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, deliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=, receivedRoutingKey=myQueue, deliveryTag=1, messageCount=0])
15:41:10.571 [SimpleAsyncTaskExecutor-1] DEBUG o.s.a.r.l.a.MessagingMessageListenerAdapter - Processing [GenericMessage [payload=Hello World, headers={timestamp=1435844470571, id=018f39f6-ebca-aabf-7fe3-a095e959f65d, amqp_receivedRoutingKey=myQueue, amqp_deliveryMode=PERSISTENT, amqp_consumerQueue=myQueue, amqp_consumerTag=amq.ctag-bUsK4KQN6_QHzf8DoDC_ww, amqp_contentEncoding=UTF-8, contentType=text/plain, amqp_deliveryTag=1, amqp_redelivered=false}]]
Received: Hello World
15:41:10.579 [SimpleAsyncTaskExecutor-1] DEBUG o.s.a.r.l.BlockingQueueConsumer - Retrieving delivery for Consumer: tags=[{amq.ctag-bUsK4KQN6_QHzf8DoDC_ww=myQueue}], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.1.1:5672/,1), acknowledgeMode=MANUAL local queue size=0
15:41:11.579 [SimpleAsyncTaskExecutor-1] DEBUG o.s.a.r.l.BlockingQueueConsumer - Retrieving delivery for Consumer: tags=[{amq.ctag-bUsK4KQN6_QHzf8DoDC_ww=myQueue}], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.1.1:5672/,1), acknowledgeMode=MANUAL local queue size=0
15:41:12.583 [SimpleAsyncTaskExecutor-1] DEBUG o.s.a.r.l.BlockingQueueConsumer - Retrieving delivery for Consumer: tags=[{amq.ctag-bUsK4KQN6_QHzf8DoDC_ww=myQueue}], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.1.1:5672/,1), acknowledgeMode=MANUAL local queue size=0
…

最后一条日志消息基本上每秒无限重复。

前两种方法可能是最有趣的部分；其余主要为Spring配置：

@Configuration
@EnableRabbit
public class MyTest {

    public static void main(String[] args) throws InterruptedException {
        try (ConfigurableApplicationContext appCtxt =
                new AnnotationConfigApplicationContext(MyTest.class)) {
            // send a test message
            RabbitTemplate template = appCtxt.getBean(RabbitTemplate.class);
            Queue queue = appCtxt.getBean(Queue.class);
            template.convertAndSend(queue.getName(), "Hello World");
            System.out.println("Sent: Hello World");

            // Now that the application with its message listeners is running,
            // block this thread forever; make sure, though, that the
            // application context can sanely be closed.
            appCtxt.registerShutdownHook();
            Object blockingObj = new Object();
            synchronized (blockingObj) {
                blockingObj.wait();
            }
        }
    }

    @RabbitListener(queues = "#{ @myQueue }")
    private void processHello(@Payload String msg,
            @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel)
            throws IOException {
        System.out.println("Received: " + msg);
        channel.basicAck(deliveryTag, false);
    }

    @Bean
    public RabbitTemplate rabbitTemplate() {
        return new RabbitTemplate(rabbitConnFactory());
    }

    @Bean
    public ConnectionFactory rabbitConnFactory() {
        return new CachingConnectionFactory();
    }

    @Bean
    public SimpleRabbitListenerContainerFactory
            rabbitListenerContainerFactory() {
        SimpleRabbitListenerContainerFactory result =
                new SimpleRabbitListenerContainerFactory();
        result.setConnectionFactory(rabbitConnFactory());
        result.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        return result;
    }

    @Bean
    public Queue myQueue() {
        return new Queue("myQueue", false);
    }

    @Bean
    public AmqpAdmin amqpAdmin() {
        return new RabbitAdmin(rabbitConnFactory());
    }
}

Answer 1

它不是轮询Rabbitmq；当消息从Rabbit异步到达时，它被放置在消费者的内部队列中；将消息移交给被阻塞的侦听器线程，等待到达。

您看到的调试消息是在侦听器线程等待来自rabbitmq的新消息超时之后。

您可以增加receiveTimeout来减少日志，或者只需为BlockingQueueConsumer禁用调试日志。

增加超时将使容器对容器Stop（）请求的响应性降低。

编辑：

为了回应你下面的评论...

是的，我们可以中断线程，但它比这要复杂得多。接收超时还用于在以下情况下确认消息

假设您只想每20条消息（而不是每条消息）进行一次访问。人们这样做是为了提高在大容量环境中的性能。超时也用于访问（txSize实际上是每n条消息或超时）。

现在，假设有19条消息到达，然后在60秒内没有消息，您的超时时间是30秒。

这意味着这19条消息将被长时间取消确认。在默认配置下，ack将在第19条消息到达后1秒发送。

在这个超时中确实没有什么开销（我们只是简单地返回并再次等待），所以增加它是不寻常的。

此外，当上下文关闭时容器停止，人们总是停止和启动容器。