Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2601: Add a batchReceiveTimeout (#2605)

Browse Source 
Fixes: #2601

Stop to waiting next message and execute listener when `batchReceiveTimeout` is timed out.

* Add `batchReceiveTimeout` to the `SimpleMessageListenerContainer` configuration.
This commit is contained in:
laststem
2024-01-26 00:52:13 +09:00
committed by GitHub
parent 7ad35b5746
commit 7166859fc0
5 changed files with 131 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java
View File

@@ -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<AbstractMe
private Long receiveTimeout;
private Long batchReceiveTimeout;
private Integer batchSize;
private Integer declarationRetries;
@@ -389,6 +392,18 @@ public class ListenerContainerFactoryBean extends AbstractFactoryBean<AbstractMe
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) {
this.batchReceiveTimeout = batchReceiveTimeout;
}
/**
* This property has several functions.
* <p>
@@ -552,6 +567,7 @@ public class ListenerContainerFactoryBean extends AbstractFactoryBean<AbstractMe
.acceptIfNotNull(this.consecutiveActiveTrigger, container::setConsecutiveActiveTrigger)
.acceptIfNotNull(this.consecutiveIdleTrigger, container::setConsecutiveIdleTrigger)
.acceptIfNotNull(this.receiveTimeout, container::setReceiveTimeout)
.acceptIfNotNull(this.batchReceiveTimeout, container::setBatchReceiveTimeout)
.acceptIfNotNull(this.batchSize, container::setBatchSize)
.acceptIfNotNull(this.consumerBatchEnabled, container::setConsumerBatchEnabled)
.acceptIfNotNull(this.declarationRetries, container::setDeclarationRetries)

21
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/SimpleRabbitListenerContainerFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2022 the original author or authors.
* Copyright 2014-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.
@@ -32,6 +32,7 @@ import org.springframework.amqp.utils.JavaUtils;
* @author Gary Russell
* @author Artem Bilan
* @author Dustin Schultz
* @author Jeonggi Kim
*
* @since 1.4
*/
@@ -54,6 +55,8 @@ public class SimpleRabbitListenerContainerFactory
private Long receiveTimeout;
private Long batchReceiveTimeout;
private Boolean consumerBatchEnabled;
/**
@@ -121,6 +124,19 @@ public class SimpleRabbitListenerContainerFactory
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 SimpleMessageListenerContainer#setBatchReceiveTimeout
* @see #setBatchSize(Integer)
*/
public void setBatchReceiveTimeout(Long batchReceiveTimeout) {
this.batchReceiveTimeout = batchReceiveTimeout;
}
/**
* Set to true to present a list of messages based on the {@link #setBatchSize(Integer)},
* if the listener supports it. Starting with version 3.0, setting this to true will
@@ -163,7 +179,8 @@ public class SimpleRabbitListenerContainerFactory
.acceptIfNotNull(this.stopConsumerMinInterval, instance::setStopConsumerMinInterval)
.acceptIfNotNull(this.consecutiveActiveTrigger, instance::setConsecutiveActiveTrigger)
.acceptIfNotNull(this.consecutiveIdleTrigger, instance::setConsecutiveIdleTrigger)
.acceptIfNotNull(this.receiveTimeout, instance::setReceiveTimeout);
.acceptIfNotNull(this.receiveTimeout, instance::setReceiveTimeout)
.acceptIfNotNull(this.batchReceiveTimeout, instance::setBatchReceiveTimeout);
if (Boolean.TRUE.equals(this.consumerBatchEnabled)) {
instance.setConsumerBatchEnabled(true);
/*

30
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2023 the original author or authors.
* Copyright 2002-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.
@@ -80,6 +80,7 @@ import com.rabbitmq.client.ShutdownSignalException;
* @author Mat Jaggard
* @author Yansong Ren
* @author Tim Bourquin
* @author Jeonggi Kim
*
* @since 1.0
*/
@@ -121,6 +122,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
private long batchReceiveTimeout;
private Set<BlockingQueueConsumer> 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.
* <p>
@@ -996,8 +1012,18 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
List<Message> 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);

54
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java
View File

@@ -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> 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<Object> messageToConsumer(final Channel mockChannel, final SimpleMessageListenerContainer container,
final boolean cancel, final CountDownLatch latch) {
return invocation -> {

14
src/reference/antora/modules/ROOT/pages/amqp/containerAttributes.adoc
View File

@@ -198,7 +198,7 @@ a|
|[[consumerBatchEnabled]]<<consumerBatchEnabled,`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]]<<batchReceiveTimeout,`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]]<<recoveryBackOff,`recoveryBackOff`>> +
(recovery-back-off)