diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java index a23d29ca..0a070e1e 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2022 the original author or authors. + * Copyright 2016-2024 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -55,6 +55,7 @@ import org.springframework.util.backoff.BackOff; * @author Gary Russell * @author Artem Bilan * @author Johno Crawford + * @author Jeonggi Kim * * @since 2.0 * @@ -166,6 +167,8 @@ public class ListenerContainerFactoryBean extends AbstractFactoryBean @@ -552,6 +567,7 @@ public class ListenerContainerFactoryBean extends AbstractFactoryBean consumers; private Integer declarationRetries; @@ -330,6 +333,19 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta this.receiveTimeout = receiveTimeout; } + /** + * The number of milliseconds of timeout for gathering batch messages. + * It limits the time to wait to fill batchSize. + * Default is 0 (no timeout). + * @param batchReceiveTimeout the timeout for gathering batch messages. + * @since 3.1.2 + * @see #setBatchSize(int) + */ + public void setBatchReceiveTimeout(long batchReceiveTimeout) { + Assert.isTrue(batchReceiveTimeout >= 0, "'batchReceiveTimeout' must be >= 0"); + this.batchReceiveTimeout = batchReceiveTimeout; + } + /** * This property has several functions. *

@@ -996,8 +1012,18 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta List messages = null; long deliveryTag = 0; - + boolean isBatchReceiveTimeoutEnabled = this.batchReceiveTimeout > 0; + long startTime = isBatchReceiveTimeoutEnabled ? System.currentTimeMillis() : 0; for (int i = 0; i < this.batchSize; i++) { + boolean batchTimedOut = isBatchReceiveTimeoutEnabled && + (System.currentTimeMillis() - startTime) > this.batchReceiveTimeout; + if (batchTimedOut) { + if (logger.isTraceEnabled()) { + long gathered = messages != null ? messages.size() : 0; + logger.trace("Timed out for gathering batch messages. gathered size is " + gathered); + } + break; + } logger.trace("Waiting for message from consumer."); Message message = consumer.nextMessage(this.receiveTimeout); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java index 7044f5ee..26c72b14 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java @@ -110,6 +110,7 @@ import com.rabbitmq.client.PossibleAuthenticationFailureException; * @author Mohammad Hewedy * @author Yansong Ren * @author Tim Bourquin + * @author Jeonggi Kim */ public class SimpleMessageListenerContainerTests { @@ -784,6 +785,59 @@ public class SimpleMessageListenerContainerTests { assertThat(start.getCount()).isEqualTo(0L); } + @Test + public void testBatchReceiveTimedOut() throws Exception { + ConnectionFactory connectionFactory = mock(ConnectionFactory.class); + Connection connection = mock(Connection.class); + Channel channel = mock(Channel.class); + given(connectionFactory.createConnection()).willReturn(connection); + given(connection.createChannel(false)).willReturn(channel); + final AtomicReference consumer = new AtomicReference<>(); + willAnswer(invocation -> { + consumer.set(invocation.getArgument(6)); + consumer.get().handleConsumeOk("1"); + return "1"; + }).given(channel) + .basicConsume(anyString(), anyBoolean(), anyString(), anyBoolean(), anyBoolean(), anyMap(), + any(Consumer.class)); + final CountDownLatch latch = new CountDownLatch(2); + willAnswer(invocation -> { + latch.countDown(); + return null; + }).given(channel).basicAck(anyLong(), anyBoolean()); + + final SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory); + container.setAfterReceivePostProcessors(msg -> null); + container.setQueueNames("foo"); + MessageListener listener = mock(BatchMessageListener.class); + container.setMessageListener(listener); + container.setBatchSize(3); + container.setConsumerBatchEnabled(true); + container.setReceiveTimeout(10); + container.setBatchReceiveTimeout(20); + container.start(); + + BasicProperties props = new BasicProperties(); + byte[] payload = "baz".getBytes(); + Envelope envelope = new Envelope(1L, false, "foo", "bar"); + consumer.get().handleDelivery("1", envelope, props, payload); + envelope = new Envelope(2L, false, "foo", "bar"); + consumer.get().handleDelivery("1", envelope, props, payload); + // waiting for batch receive timed out + Thread.sleep(20); + envelope = new Envelope(3L, false, "foo", "bar"); + consumer.get().handleDelivery("1", envelope, props, payload); + assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue(); + verify(channel, never()).basicAck(eq(1), anyBoolean()); + verify(channel).basicAck(2, true); + verify(channel, never()).basicAck(eq(2), anyBoolean()); + verify(channel).basicAck(3, true); + container.stop(); + verify(listener).containerAckMode(AcknowledgeMode.AUTO); + verify(listener).isAsyncReplies(); + verifyNoMoreInteractions(listener); + } + private Answer messageToConsumer(final Channel mockChannel, final SimpleMessageListenerContainer container, final boolean cancel, final CountDownLatch latch) { return invocation -> { diff --git a/src/reference/antora/modules/ROOT/pages/amqp/containerAttributes.adoc b/src/reference/antora/modules/ROOT/pages/amqp/containerAttributes.adoc index 4f9d827f..9aec7d7c 100644 --- a/src/reference/antora/modules/ROOT/pages/amqp/containerAttributes.adoc +++ b/src/reference/antora/modules/ROOT/pages/amqp/containerAttributes.adoc @@ -198,7 +198,7 @@ a| |[[consumerBatchEnabled]]<> + (batch-enabled) -|If the `MessageListener` supports it, setting this to true enables batching of discrete messages, up to `batchSize`; a partial batch will be delivered if no new messages arrive in `receiveTimeout`. +|If the `MessageListener` supports it, setting this to true enables batching of discrete messages, up to `batchSize`; a partial batch will be delivered if no new messages arrive in `receiveTimeout` or gathering batch messages time exceeded `batchReceiveTimeout`. When this is false, batching is only supported for batches created by a producer; see xref:amqp/sending-messages.adoc#template-batching[Batching]. a|image::tickmark.png[] @@ -611,6 +611,18 @@ a|image::tickmark.png[] a| a| +|[[batchReceiveTimeout]]<> + +(batch-receive-timeout) + +|The number of milliseconds of timeout for gathering batch messages. +It limits the time to wait to fill batchSize. +When `batchSize > 1` and the time to gathering batch messages is greater than `batchReceiveTime`, batch will be delivered. +Default is 0 (no timeout). + +a|image::tickmark.png[] +a| +a| + |[[recoveryBackOff]]<> + (recovery-back-off)