GH-922: Add Batch-mode @RabbitListener

Resolves https://github.com/spring-projects/spring-amqp/issues/922

- remove unnecessary back-ticks from literal table columns

* Polishing - PR Comments

spring-amqp/src/main/java/org/springframework/amqp/support/converter/MessagingMessageConverter.java

@@ -88,6 +88,10 @@ public class MessagingMessageConverter implements MessageConverter, Initializing
 		this.headerMapper = headerMapper;
 	}
 
+	public AmqpHeaderMapper getHeaderMapper() {
+		return this.headerMapper;
+	}
+
 	@Override
 	public void afterPropertiesSet() {
 		Assert.notNull(this.payloadConverter, "Property 'payloadConverter' is required");

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java

@@ -29,6 +29,7 @@ import org.apache.commons.logging.LogFactory;
 import org.springframework.amqp.core.AcknowledgeMode;
 import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.support.BatchingStrategy;
 import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
 import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
@@ -119,6 +120,10 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
 
 	private Consumer<C> containerConfigurer;
 
+	private boolean batchListener;
+
+	private BatchingStrategy batchingStrategy;
+
 	/**
 	 * @param connectionFactory The connection factory.
 	 * @see AbstractMessageListenerContainer#setConnectionFactory(ConnectionFactory)
@@ -344,6 +349,27 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
 		this.containerConfigurer = configurer;
 	}
 
+	/**
+	 * Set to true to receive a list of debatched messages that were created by a
+	 * {@link org.springframework.amqp.rabbit.core.BatchingRabbitTemplate}.
+	 * @param isBatch true for a batch listener.
+	 * @since 2.2
+	 * @see #setBatchingStrategy(BatchingStrategy)
+	 */
+	public void setBatchListener(boolean isBatch) {
+		this.batchListener = isBatch;
+	}
+
+	/**
+	 * Set a {@link BatchingStrategy} to use when debatching messages.
+	 * @param batchingStrategy the batching strategy.
+	 * @since 2.2
+	 * @see #setBatchListener(boolean)
+	 */
+	public void setBatchingStrategy(BatchingStrategy batchingStrategy) {
+		this.batchingStrategy = batchingStrategy;
+	}
+
 	@Override
 	public C createListenerContainer(RabbitListenerEndpoint endpoint) {
 		C instance = createContainerInstance();
@@ -356,30 +382,33 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
 			endpoint.setMessageConverter(this.messageConverter);
 		}
 		javaUtils
-				.acceptIfNotNull(this.acknowledgeMode, instance::setAcknowledgeMode)
-				.acceptIfNotNull(this.channelTransacted, instance::setChannelTransacted)
-				.acceptIfNotNull(this.applicationContext, instance::setApplicationContext)
-				.acceptIfNotNull(this.taskExecutor, instance::setTaskExecutor)
-				.acceptIfNotNull(this.transactionManager, instance::setTransactionManager)
-				.acceptIfNotNull(this.prefetchCount, instance::setPrefetchCount)
-				.acceptIfNotNull(this.defaultRequeueRejected, instance::setDefaultRequeueRejected)
-				.acceptIfNotNull(this.adviceChain, instance::setAdviceChain)
-				.acceptIfNotNull(this.recoveryBackOff, instance::setRecoveryBackOff)
-				.acceptIfNotNull(this.mismatchedQueuesFatal, instance::setMismatchedQueuesFatal)
-				.acceptIfNotNull(this.missingQueuesFatal, instance::setMissingQueuesFatal)
-				.acceptIfNotNull(this.consumerTagStrategy, instance::setConsumerTagStrategy)
-				.acceptIfNotNull(this.idleEventInterval, instance::setIdleEventInterval)
-				.acceptIfNotNull(this.failedDeclarationRetryInterval, instance::setFailedDeclarationRetryInterval)
-				.acceptIfNotNull(this.applicationEventPublisher, instance::setApplicationEventPublisher)
-				.acceptIfNotNull(this.autoStartup, instance::setAutoStartup)
-				.acceptIfNotNull(this.phase, instance::setPhase)
-				.acceptIfNotNull(this.afterReceivePostProcessors, instance::setAfterReceivePostProcessors);
+			.acceptIfNotNull(this.acknowledgeMode, instance::setAcknowledgeMode)
+			.acceptIfNotNull(this.channelTransacted, instance::setChannelTransacted)
+			.acceptIfNotNull(this.applicationContext, instance::setApplicationContext)
+			.acceptIfNotNull(this.taskExecutor, instance::setTaskExecutor)
+			.acceptIfNotNull(this.transactionManager, instance::setTransactionManager)
+			.acceptIfNotNull(this.prefetchCount, instance::setPrefetchCount)
+			.acceptIfNotNull(this.defaultRequeueRejected, instance::setDefaultRequeueRejected)
+			.acceptIfNotNull(this.adviceChain, instance::setAdviceChain)
+			.acceptIfNotNull(this.recoveryBackOff, instance::setRecoveryBackOff)
+			.acceptIfNotNull(this.mismatchedQueuesFatal, instance::setMismatchedQueuesFatal)
+			.acceptIfNotNull(this.missingQueuesFatal, instance::setMissingQueuesFatal)
+			.acceptIfNotNull(this.consumerTagStrategy, instance::setConsumerTagStrategy)
+			.acceptIfNotNull(this.idleEventInterval, instance::setIdleEventInterval)
+			.acceptIfNotNull(this.failedDeclarationRetryInterval, instance::setFailedDeclarationRetryInterval)
+			.acceptIfNotNull(this.applicationEventPublisher, instance::setApplicationEventPublisher)
+			.acceptIfNotNull(this.autoStartup, instance::setAutoStartup)
+			.acceptIfNotNull(this.phase, instance::setPhase)
+			.acceptIfNotNull(this.afterReceivePostProcessors, instance::setAfterReceivePostProcessors);
+		instance.setDeBatchingEnabled(!this.batchListener);
 		if (endpoint != null) { // endpoint settings overriding default factory settings
 			javaUtils
-					.acceptIfNotNull(endpoint.getAutoStartup(), instance::setAutoStartup)
-					.acceptIfNotNull(endpoint.getTaskExecutor(), instance::setTaskExecutor);
+				.acceptIfNotNull(endpoint.getAutoStartup(), instance::setAutoStartup)
+				.acceptIfNotNull(endpoint.getTaskExecutor(), instance::setTaskExecutor);
+			javaUtils
+				.acceptIfNotNull(this.batchingStrategy, endpoint::setBatchingStrategy);
 			instance.setListenerId(endpoint.getId());
-
+			endpoint.setBatchListener(this.batchListener);
 			endpoint.setupListenerContainer(instance);
 		}
 		if (instance.getMessageListener() instanceof AbstractAdaptableMessageListener) {

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java

@@ -363,7 +363,7 @@ public abstract class RabbitUtils {
 				return rcon.getFrameMax();
 			}
 		}
-		catch (RuntimeException e) {
+		catch (@SuppressWarnings("unused") RuntimeException e) {
 			// NOSONAR
 		}
 		return -1;

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplate.java

@@ -21,9 +21,11 @@ import java.util.concurrent.ScheduledFuture;
 
 import org.springframework.amqp.AmqpException;
 import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.connection.CorrelationData;
 import org.springframework.amqp.rabbit.core.support.BatchingStrategy;
 import org.springframework.amqp.rabbit.core.support.MessageBatch;
+import org.springframework.lang.Nullable;
 import org.springframework.scheduling.TaskScheduler;
 
 /**
@@ -48,10 +50,25 @@ public class BatchingRabbitTemplate extends RabbitTemplate {
 	private volatile ScheduledFuture<?> scheduledTask;
 
 	/**
+	 * Create an instance with the supplied parameters.
 	 * @param batchingStrategy the batching strategy.
 	 * @param scheduler the scheduler.
 	 */
 	public BatchingRabbitTemplate(BatchingStrategy batchingStrategy, TaskScheduler scheduler) {
+		this(null, batchingStrategy, scheduler);
+	}
+
+	/**
+	 * Create an instance with the supplied parameters.
+	 * @param connectionFactory the connection factory.
+	 * @param batchingStrategy the batching strategy.
+	 * @param scheduler the scheduler.
+	 * @since 2.2
+	 */
+	public BatchingRabbitTemplate(@Nullable ConnectionFactory connectionFactory, BatchingStrategy batchingStrategy,
+			TaskScheduler scheduler) {
+
+		super(connectionFactory);
 		this.batchingStrategy = batchingStrategy;
 		this.scheduler = scheduler;
 	}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/support/BatchingStrategy.java

@@ -18,8 +18,10 @@ package org.springframework.amqp.rabbit.core.support;
 
 import java.util.Collection;
 import java.util.Date;
+import java.util.function.Consumer;
 
 import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
 
 /**
  * Strategy for batching messages. The methods will never be called concurrently.
@@ -52,4 +54,27 @@ public interface BatchingStrategy {
 	 */
 	Collection<MessageBatch> releaseBatches();
 
+	/**
+	 * Return true if this strategy can decode a batch of messages from a message body.
+	 * Returning true means you must override {@link #deBatch(Message, Consumer)}.
+	 * @param properties the message properties.
+	 * @return true if we can decode the message.
+	 * @since 2.2
+	 * @see #deBatch(Message, Consumer)
+	 */
+	default boolean canDebatch(MessageProperties properties) {
+		return false;
+	}
+
+	/**
+	 * Decode a message into fragments.
+	 * @param message the message.
+	 * @param fragmentConsumer a consumer for fragments.
+	 * @since 2.2
+	 * @see #canDebatch(MessageProperties)
+	 */
+	default void deBatch(Message message, Consumer<Message> fragmentConsumer) {
+		throw new UnsupportedOperationException("Cannot debatch this message");
+	}
+
 }

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/support/SimpleBatchingStrategy.java

@@ -22,9 +22,12 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.Date;
 import java.util.List;
+import java.util.function.Consumer;
 
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
+import org.springframework.amqp.support.converter.MessageConversionException;
 import org.springframework.util.Assert;
 
 /**
@@ -66,25 +69,26 @@ public class SimpleBatchingStrategy implements BatchingStrategy {
 	}
 
 	@Override
-	public MessageBatch addToBatch(String exchange, String routingKey, Message message) {
+	public MessageBatch addToBatch(String exch, String routKey, Message message) {
 		if (this.exchange != null) {
-			Assert.isTrue(this.exchange.equals(exchange), "Cannot send to different exchanges in the same batch");
+			Assert.isTrue(this.exchange.equals(exch), "Cannot send to different exchanges in the same batch");
 		}
 		else {
-			this.exchange = exchange;
+			this.exchange = exch;
 		}
 		if (this.routingKey != null) {
-			Assert.isTrue(this.routingKey.equals(routingKey), "Cannot send with different routing keys in the same batch");
+			Assert.isTrue(this.routingKey.equals(routKey),
+					"Cannot send with different routing keys in the same batch");
 		}
 		else {
-			this.routingKey = routingKey;
+			this.routingKey = routKey;
 		}
 		int bufferUse = Integer.BYTES + message.getBody().length;
 		MessageBatch batch = null;
 		if (this.messages.size() > 0 && this.currentSize + bufferUse > this.bufferLimit) {
 			batch = doReleaseBatch();
-			this.exchange = exchange;
-			this.routingKey = routingKey;
+			this.exchange = exch;
+			this.routingKey = routKey;
 		}
 		this.currentSize += bufferUse;
 		this.messages.add(message);
@@ -144,8 +148,45 @@ public class SimpleBatchingStrategy implements BatchingStrategy {
 			bytes.putInt(message.getBody().length);
 			bytes.put(message.getBody());
 		}
-		messageProperties.getHeaders().put(MessageProperties.SPRING_BATCH_FORMAT, MessageProperties.BATCH_FORMAT_LENGTH_HEADER4);
+		messageProperties.getHeaders().put(MessageProperties.SPRING_BATCH_FORMAT,
+				MessageProperties.BATCH_FORMAT_LENGTH_HEADER4);
 		return new Message(body, messageProperties);
 	}
 
+
+	@Override
+	public boolean canDebatch(MessageProperties properties) {
+		return MessageProperties.BATCH_FORMAT_LENGTH_HEADER4.equals(properties
+				.getHeaders()
+				.get(MessageProperties.SPRING_BATCH_FORMAT));
+	}
+
+	/**
+	 * Debatch a message that has a header with {@link MessageProperties#SPRING_BATCH_FORMAT}
+	 * set to {@link MessageProperties#BATCH_FORMAT_LENGTH_HEADER4}.
+	 * @param message the batched message.
+	 * @param fragmentConsumer a consumer for each fragment.
+	 * @since 2.2
+	 */
+	@Override
+	public void deBatch(Message message, Consumer<Message> fragmentConsumer) {
+		ByteBuffer byteBuffer = ByteBuffer.wrap(message.getBody());
+		MessageProperties messageProperties = message.getMessageProperties();
+		messageProperties.getHeaders().remove(MessageProperties.SPRING_BATCH_FORMAT);
+		while (byteBuffer.hasRemaining()) {
+			int length = byteBuffer.getInt();
+			if (length < 0 || length > byteBuffer.remaining()) {
+				throw new ListenerExecutionFailedException("Bad batched message received",
+						new MessageConversionException("Insufficient batch data at offset " + byteBuffer.position()),
+						message);
+			}
+			byte[] body = new byte[length];
+			byteBuffer.get(body);
+			messageProperties.setContentLength(length);
+			// Caveat - shared MessageProperties.
+			Message fragment = new Message(body, messageProperties);
+			fragmentConsumer.accept(fragment);
+		}
+	}
+
 }

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java

@@ -16,7 +16,6 @@
 
 package org.springframework.amqp.rabbit.listener;
 
-import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -44,7 +43,6 @@ import org.springframework.amqp.core.AmqpAdmin;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageListener;
 import org.springframework.amqp.core.MessagePostProcessor;
-import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.Connection;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
@@ -53,6 +51,8 @@ import org.springframework.amqp.rabbit.connection.RabbitAccessor;
 import org.springframework.amqp.rabbit.connection.RabbitResourceHolder;
 import org.springframework.amqp.rabbit.connection.RabbitUtils;
 import org.springframework.amqp.rabbit.connection.RoutingConnectionFactory;
+import org.springframework.amqp.rabbit.core.support.BatchingStrategy;
+import org.springframework.amqp.rabbit.core.support.SimpleBatchingStrategy;
 import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
 import org.springframework.amqp.rabbit.listener.exception.FatalListenerExecutionException;
 import org.springframework.amqp.rabbit.listener.exception.FatalListenerStartupException;
@@ -61,7 +61,6 @@ import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter
 import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
 import org.springframework.amqp.support.ConditionalExceptionLogger;
 import org.springframework.amqp.support.ConsumerTagStrategy;
-import org.springframework.amqp.support.converter.MessageConversionException;
 import org.springframework.amqp.support.postprocessor.MessagePostProcessorUtils;
 import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.aop.support.DefaultPointcutAdvisor;
@@ -219,6 +218,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 
 	private String errorHandlerLoggerName = getClass().getName();
 
+	private BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(0, 0, 0L);
+
 	private volatile boolean lazyLoad;
 
 	@Override
@@ -440,6 +441,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 	 * Determine whether or not the container should de-batch batched
 	 * messages (true) or call the listener with the batch (false). Default: true.
 	 * @param deBatchingEnabled the deBatchingEnabled to set.
+	 * @see #setBatchingStrategy(BatchingStrategy)
 	 */
 	public void setDeBatchingEnabled(boolean deBatchingEnabled) {
 		this.deBatchingEnabled = deBatchingEnabled;
@@ -1045,6 +1047,18 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 		this.errorHandlerLoggerName = errorHandlerLoggerName;
 	}
 
+	/**
+	 * Set a batching strategy to use when de-batching messages.
+	 * Default is {@link SimpleBatchingStrategy}.
+	 * @param batchingStrategy the strategy.
+	 * @since 2.2
+	 * @see #setDeBatchingEnabled(boolean)
+	 */
+	public void setBatchingStrategy(BatchingStrategy batchingStrategy) {
+		Assert.notNull(batchingStrategy, "'batchingStrategy' cannot be null");
+		this.batchingStrategy = batchingStrategy;
+	}
+
 	/**
 	 * Delegates to {@link #validateConfiguration()} and {@link #initialize()}.
 	 */
@@ -1359,25 +1373,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 				}
 			}
 		}
-		Object batchFormat = message.getMessageProperties().getHeaders().get(MessageProperties.SPRING_BATCH_FORMAT);
-		if (MessageProperties.BATCH_FORMAT_LENGTH_HEADER4.equals(batchFormat) && this.deBatchingEnabled) {
-			ByteBuffer byteBuffer = ByteBuffer.wrap(message.getBody());
-			MessageProperties messageProperties = message.getMessageProperties();
-			messageProperties.getHeaders().remove(MessageProperties.SPRING_BATCH_FORMAT);
-			while (byteBuffer.hasRemaining()) {
-				int length = byteBuffer.getInt();
-				if (length < 0 || length > byteBuffer.remaining()) {
-					throw new ListenerExecutionFailedException("Bad batched message received",
-							new MessageConversionException("Insufficient batch data at offset " + byteBuffer.position()),
-							message);
-				}
-				byte[] body = new byte[length];
-				byteBuffer.get(body);
-				messageProperties.setContentLength(length);
-				// Caveat - shared MessageProperties.
-				Message fragment = new Message(body, messageProperties);
-				invokeListener(channel, fragment);
-			}
+		if (this.deBatchingEnabled && this.batchingStrategy.canDebatch(message.getMessageProperties())) {
+			this.batchingStrategy.deBatch(message, fragment -> invokeListener(channel, fragment));
 		}
 		else {
 			invokeListener(channel, message);

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractRabbitListenerEndpoint.java

@@ -25,6 +25,7 @@ import java.util.Map;
 import org.springframework.amqp.core.AmqpAdmin;
 import org.springframework.amqp.core.MessageListener;
 import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.core.support.BatchingStrategy;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
@@ -82,6 +83,10 @@ public abstract class AbstractRabbitListenerEndpoint implements RabbitListenerEn
 
 	private TaskExecutor taskExecutor;
 
+	private boolean batchListener;
+
+	private BatchingStrategy batchingStrategy;
+
 	@Override
 	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
 		this.beanFactory = beanFactory;
@@ -279,6 +284,31 @@ public abstract class AbstractRabbitListenerEndpoint implements RabbitListenerEn
 		this.taskExecutor = taskExecutor;
 	}
 
+	public boolean isBatchListener() {
+		return this.batchListener;
+	}
+
+	/**
+	 * Set to true if this endpoint should create a batch listener.
+	 * @param batchListener true for a batch listener.
+	 * @since 2.2
+	 * @see #setBatchingStrategy(BatchingStrategy)
+	 */
+	@Override
+	public void setBatchListener(boolean batchListener) {
+		this.batchListener = batchListener;
+	}
+
+	@Nullable
+	public BatchingStrategy getBatchingStrategy() {
+		return this.batchingStrategy;
+	}
+
+	@Override
+	public void setBatchingStrategy(BatchingStrategy batchingStrategy) {
+		this.batchingStrategy = batchingStrategy;
+	}
+
 	@Override
 	public void setupListenerContainer(MessageListenerContainer listenerContainer) {
 		AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) listenerContainer;

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpoint.java

@@ -19,6 +19,7 @@ package org.springframework.amqp.rabbit.listener;
 import java.lang.reflect.Method;
 import java.util.Arrays;
 
+import org.springframework.amqp.rabbit.listener.adapter.BatchMessagingMessageListenerAdapter;
 import org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter;
 import org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter;
 import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
@@ -150,7 +151,14 @@ public class MethodRabbitListenerEndpoint extends AbstractRabbitListenerEndpoint
 	 * @return the {@link MessagingMessageListenerAdapter} instance.
 	 */
 	protected MessagingMessageListenerAdapter createMessageListenerInstance() {
-		return new MessagingMessageListenerAdapter(this.bean, this.method, this.returnExceptions, this.errorHandler);
+		if (isBatchListener()) {
+			return new BatchMessagingMessageListenerAdapter(this.bean, this.method, this.returnExceptions,
+					this.errorHandler, getBatchingStrategy());
+		}
+		else {
+			return new MessagingMessageListenerAdapter(this.bean, this.method, this.returnExceptions,
+					this.errorHandler);
+		}
 	}
 
 	@Nullable

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/RabbitListenerEndpoint.java

@@ -16,6 +16,7 @@
 
 package org.springframework.amqp.rabbit.listener;
 
+import org.springframework.amqp.rabbit.core.support.BatchingStrategy;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.lang.Nullable;
@@ -105,4 +106,23 @@ public interface RabbitListenerEndpoint {
 		return null;
 	}
 
+	/**
+	 * Called by the container factory with the factory's batchListener property.
+	 * @param batchListener the batchListener to set.
+	 * @since 2.2
+	 */
+	default void setBatchListener(boolean batchListener) {
+		// NOSONAR empty
+	}
+
+	/**
+	 * Set a {@link BatchingStrategy} to use when debatching messages.
+	 * @param batchingStrategy the batching strategy.
+	 * @since 2.2
+	 * @see #setBatchListener(boolean)
+	 */
+	default void setBatchingStrategy(BatchingStrategy batchingStrategy) {
+		// NOSONAR empty
+	}
+
 }

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/BatchMessagingMessageListenerAdapter.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright 2019 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.amqp.rabbit.listener.adapter;
+
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.amqp.rabbit.core.support.BatchingStrategy;
+import org.springframework.amqp.rabbit.core.support.SimpleBatchingStrategy;
+import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.messaging.support.MessageBuilder;
+
+/**
+ * A listener adapter for batch listeners.
+ *
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public class BatchMessagingMessageListenerAdapter extends MessagingMessageListenerAdapter {
+
+	private final MessagingMessageConverterAdapter converterAdapter;
+
+	private final BatchingStrategy batchingStrategy;
+
+	public BatchMessagingMessageListenerAdapter(Object bean, Method method, boolean returnExceptions,
+			RabbitListenerErrorHandler errorHandler, @Nullable BatchingStrategy batchingStrategy) {
+
+		super(bean, method, returnExceptions, errorHandler, true);
+		this.converterAdapter = (MessagingMessageConverterAdapter) getMessagingMessageConverter();
+		this.batchingStrategy = batchingStrategy == null ? new SimpleBatchingStrategy(0, 0, 0L) : batchingStrategy;
+	}
+
+	@Override
+	protected Message<?> toMessagingMessage(org.springframework.amqp.core.Message amqpMessage) {
+		if (this.batchingStrategy.canDebatch(amqpMessage.getMessageProperties())) {
+
+			if (this.converterAdapter.isMessageList()) {
+				List<Message<?>> messages = new ArrayList<>();
+				this.batchingStrategy.deBatch(amqpMessage, fragment -> {
+					messages.add(super.toMessagingMessage(fragment));
+				});
+				return new GenericMessage<>(messages);
+			}
+			else {
+				List<Object> list = new ArrayList<>();
+				this.batchingStrategy.deBatch(amqpMessage, fragment -> {
+					list.add(this.converterAdapter.extractPayload(fragment));
+				});
+				return MessageBuilder.withPayload(list)
+						.copyHeaders(this.converterAdapter
+								.getHeaderMapper()
+								.toHeaders(amqpMessage.getMessageProperties()))
+						.build();
+			}
+		}
+		return super.toMessagingMessage(amqpMessage);
+	}
+
+}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java

@@ -20,6 +20,7 @@ import java.lang.reflect.Method;
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
 import java.lang.reflect.WildcardType;
+import java.util.List;
 
 import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
@@ -76,7 +77,13 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis
 
 	public MessagingMessageListenerAdapter(Object bean, Method method, boolean returnExceptions,
 			RabbitListenerErrorHandler errorHandler) {
-		this.messagingMessageConverter = new MessagingMessageConverterAdapter(bean, method);
+		this(bean, method, returnExceptions, errorHandler, false);
+	}
+
+	protected MessagingMessageListenerAdapter(Object bean, Method method, boolean returnExceptions,
+			RabbitListenerErrorHandler errorHandler, boolean batch) {
+
+		this.messagingMessageConverter = new MessagingMessageConverterAdapter(bean, method, batch);
 		this.returnExceptions = returnExceptions;
 		this.errorHandler = errorHandler;
 	}
@@ -242,7 +249,7 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis
 	 * If the inbound message has no type information and the configured message converter
 	 * supports it, we attempt to infer the conversion type from the method signature.
 	 */
-	private final class MessagingMessageConverterAdapter extends MessagingMessageConverter {
+	protected final class MessagingMessageConverterAdapter extends MessagingMessageConverter {
 
 		private final Object bean;
 
@@ -250,15 +257,24 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis
 
 		private final Type inferredArgumentType;
 
-		MessagingMessageConverterAdapter(Object bean, Method method) {
+		private final boolean isBatch;
+
+		private boolean isMessageList;
+
+		MessagingMessageConverterAdapter(Object bean, Method method, boolean batch) {
 			this.bean = bean;
 			this.method = method;
+			this.isBatch = batch;
 			this.inferredArgumentType = determineInferredType();
 			if (logger.isDebugEnabled() && this.inferredArgumentType != null) {
 				logger.debug("Inferred argument type for " + method.toString() + " is " + this.inferredArgumentType);
 			}
 		}
 
+		protected boolean isMessageList() {
+			return this.isMessageList;
+		}
+
 		@Override
 		protected Object extractPayload(org.springframework.amqp.core.Message message) {
 			MessageProperties messageProperties = message.getMessageProperties();
@@ -291,14 +307,7 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis
 						&& (methodParameter.getParameterAnnotations().length == 0
 						|| methodParameter.hasParameterAnnotation(Payload.class))) {
 					if (genericParameterType == null) {
-						genericParameterType = methodParameter.getGenericParameterType();
-						if (genericParameterType instanceof ParameterizedType) {
-							ParameterizedType parameterizedType = (ParameterizedType) genericParameterType;
-							if (parameterizedType.getRawType().equals(Message.class)) {
-								genericParameterType = ((ParameterizedType) genericParameterType)
-									.getActualTypeArguments()[0];
-							}
-						}
+						genericParameterType = extractGenericParameterTypFromMethodParameter(methodParameter);
 					}
 					else {
 						if (MessagingMessageListenerAdapter.this.logger.isDebugEnabled()) {
@@ -333,6 +342,31 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis
 			return !parameterType.equals(Message.class); // could be Message without a generic type
 		}
 
+		private Type extractGenericParameterTypFromMethodParameter(MethodParameter methodParameter) {
+			Type genericParameterType = methodParameter.getGenericParameterType();
+			if (genericParameterType instanceof ParameterizedType) {
+				ParameterizedType parameterizedType = (ParameterizedType) genericParameterType;
+				if (parameterizedType.getRawType().equals(Message.class)) {
+					genericParameterType = ((ParameterizedType) genericParameterType).getActualTypeArguments()[0];
+				}
+				else if (parameterizedType.getRawType().equals(List.class)
+						&& parameterizedType.getActualTypeArguments().length == 1) {
+
+					Type paramType = parameterizedType.getActualTypeArguments()[0];
+					boolean messageHasGeneric = paramType instanceof ParameterizedType
+							&& ((ParameterizedType) paramType).getRawType().equals(Message.class);
+					this.isMessageList = paramType.equals(Message.class) || messageHasGeneric;
+					if (messageHasGeneric) {
+						genericParameterType = ((ParameterizedType) paramType).getActualTypeArguments()[0];
+					}
+					if (this.isBatch) {
+						// when decoding batch messages we convert to the List's generic type
+						genericParameterType = paramType;
+					}
+				}
+			}
+			return genericParameterType;
+		}
 	}
 
 }

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitBatchIntegrationTests.java

@@ -0,0 +1,156 @@
+/*
+ * Copyright 2019 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.amqp.rabbit.annotation;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate;
+import org.springframework.amqp.rabbit.core.support.SimpleBatchingStrategy;
+import org.springframework.amqp.rabbit.junit.RabbitAvailable;
+import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.Message;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+/**
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+@SpringJUnitConfig
+@DirtiesContext
+@RabbitAvailable(queues = { "batch.1", "batch.2" })
+public class EnableRabbitBatchIntegrationTests {
+
+	@Autowired
+	private BatchingRabbitTemplate template;
+
+	@Autowired
+	private Listener listener;
+
+	@Test
+	public void simpleList() throws InterruptedException {
+		this.template.convertAndSend("batch.1", new Foo("foo"));
+		this.template.convertAndSend("batch.1", new Foo("bar"));
+		assertThat(this.listener.foosLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.listener.foos.get(0).getBar()).isEqualTo("foo");
+		assertThat(this.listener.foos.get(1).getBar()).isEqualTo("bar");
+	}
+
+	@Test
+	public void messageList() throws InterruptedException {
+		this.template.convertAndSend("batch.2", new Foo("foo"));
+		this.template.convertAndSend("batch.2", new Foo("bar"));
+		assertThat(this.listener.fooMessagesLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.listener.fooMessages.get(0).getPayload().getBar()).isEqualTo("foo");
+		assertThat(this.listener.fooMessages.get(1).getPayload().getBar()).isEqualTo("bar");
+	}
+
+	@Configuration
+	@EnableRabbit
+	public static class Config {
+
+		@Bean
+		public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
+			SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
+			factory.setConnectionFactory(connectionFactory());
+			factory.setBatchListener(true);
+			return factory;
+		}
+
+		@Bean
+		public BatchingRabbitTemplate template() {
+			return new BatchingRabbitTemplate(connectionFactory(), new SimpleBatchingStrategy(2, 10_000, 10_000L),
+					scheduler());
+		}
+
+		@Bean
+		public ConnectionFactory connectionFactory() {
+			return new CachingConnectionFactory(RabbitAvailableCondition.getBrokerRunning().getConnectionFactory());
+		}
+
+		@Bean
+		public TaskScheduler scheduler() {
+			return new ThreadPoolTaskScheduler();
+		}
+
+		@Bean
+		public Listener listener() {
+			return new Listener();
+		}
+
+	}
+
+	public static class Listener {
+
+		List<Foo> foos;
+
+		CountDownLatch foosLatch = new CountDownLatch(1);
+
+		List<Message<Foo>> fooMessages;
+
+		CountDownLatch fooMessagesLatch = new CountDownLatch(1);
+
+		@RabbitListener(queues = "batch.1")
+		public void listen1(List<Foo> in) {
+			this.foos = in;
+			this.foosLatch.countDown();
+		}
+
+		@RabbitListener(queues = "batch.2")
+		public void listen2(List<Message<Foo>> in) {
+			this.fooMessages = in;
+			this.fooMessagesLatch.countDown();
+		}
+
+	}
+
+	@SuppressWarnings("serial")
+	public static class Foo implements Serializable {
+
+		private String bar;
+
+		public Foo(String bar) {
+			this.bar = bar;
+		}
+
+		public String getBar() {
+			return this.bar;
+		}
+
+		public void setBar(String bar) {
+			this.bar = bar;
+		}
+
+	}
+
+}

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitBatchJsonIntegrationTests.java

@@ -0,0 +1,167 @@
+/*
+ * Copyright 2019 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.amqp.rabbit.annotation;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate;
+import org.springframework.amqp.rabbit.core.support.SimpleBatchingStrategy;
+import org.springframework.amqp.rabbit.junit.RabbitAvailable;
+import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition;
+import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.Message;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+/**
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+@SpringJUnitConfig
+@DirtiesContext
+@RabbitAvailable(queues = { "json.batch.1", "json.batch.2" })
+public class EnableRabbitBatchJsonIntegrationTests {
+
+	@Autowired
+	private BatchingRabbitTemplate template;
+
+	@Autowired
+	private Listener listener;
+
+	@Test
+	public void testSimpleList() throws InterruptedException {
+		this.template.convertAndSend("json.batch.1", new Foo("foo"));
+		this.template.convertAndSend("json.batch.1", new Foo("bar"));
+		assertThat(this.listener.foosLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.listener.foos.get(0).getBar()).isEqualTo("foo");
+		assertThat(this.listener.foos.get(1).getBar()).isEqualTo("bar");
+	}
+
+	@Test
+	public void testMessageList() throws InterruptedException {
+		this.template.convertAndSend("json.batch.2", new Foo("foo"));
+		this.template.convertAndSend("json.batch.2", new Foo("bar"));
+		assertThat(this.listener.fooMessagesLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.listener.fooMessages.get(0).getPayload().getBar()).isEqualTo("foo");
+		assertThat(this.listener.fooMessages.get(1).getPayload().getBar()).isEqualTo("bar");
+	}
+
+	@Configuration
+	@EnableRabbit
+	public static class Config {
+
+		@Bean
+		public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
+			SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
+			factory.setConnectionFactory(connectionFactory());
+			factory.setBatchListener(true);
+			factory.setMessageConverter(converter());
+			return factory;
+		}
+
+		@Bean
+		public BatchingRabbitTemplate template() {
+			BatchingRabbitTemplate batchTemplate = new BatchingRabbitTemplate(connectionFactory(),
+					new SimpleBatchingStrategy(2, 10_000, 10_000L), scheduler());
+			batchTemplate.setMessageConverter(converter());
+			return batchTemplate;
+		}
+
+		@Bean
+		public Jackson2JsonMessageConverter converter() {
+			return new Jackson2JsonMessageConverter();
+		}
+
+		@Bean
+		public ConnectionFactory connectionFactory() {
+			return new CachingConnectionFactory(RabbitAvailableCondition.getBrokerRunning().getConnectionFactory());
+		}
+
+		@Bean
+		public TaskScheduler scheduler() {
+			return new ThreadPoolTaskScheduler();
+		}
+
+		@Bean
+		public Listener listener() {
+			return new Listener();
+		}
+
+	}
+
+	public static class Listener {
+
+		List<Foo> foos;
+
+		CountDownLatch foosLatch = new CountDownLatch(1);
+
+		List<Message<Foo>> fooMessages;
+
+		CountDownLatch fooMessagesLatch = new CountDownLatch(1);
+
+		@RabbitListener(queues = "json.batch.1")
+		public void listen1(List<Foo> in) {
+			this.foos = in;
+			this.foosLatch.countDown();
+		}
+
+		@RabbitListener(queues = "json.batch.2")
+		public void listen2(List<Message<Foo>> in) {
+			this.fooMessages = in;
+			this.fooMessagesLatch.countDown();
+		}
+
+	}
+
+	public static class Foo {
+
+		private String bar;
+
+		public Foo() {
+			super();
+		}
+
+		public Foo(String bar) {
+			this.bar = bar;
+		}
+
+		public String getBar() {
+			return this.bar;
+		}
+
+		public void setBar(String bar) {
+			this.bar = bar;
+		}
+
+	}
+
+}

src/reference/asciidoc/amqp.adoc

@@ -742,32 +742,32 @@ The following table describes the `CacheMode.CHANNEL` properties:
 
 | Meaning
 
-| `connectionName`
+| connectionName
 
 | The name of the connection generated by the `ConnectionNameStrategy`.
 
-| `channelCacheSize`
+| channelCacheSize
 
 | The currently configured maximum channels that are allowed to be idle.
 
-| `localPort`
+| localPort
 
 | The local port for the connection (if available).
 This can be used to correlate with connections and channels on the RabbitMQ Admin UI.
 
-| `idleChannelsTx`
+| idleChannelsTx
 
 | The number of transactional channels that are currently idle (cached).
 
-| `idleChannelsNotTx`
+| idleChannelsNotTx
 
 | The number of non-transactional channels that are currently idle (cached).
 
-| `idleChannelsTxHighWater`
+| idleChannelsTxHighWater
 
 | The maximum number of transactional channels that have been concurrently idle (cached).
 
-| `idleChannelsNotTxHighWater`
+| idleChannelsNotTxHighWater
 
 | The maximum number of non-transactional channels have been concurrently idle (cached).
 
@@ -782,46 +782,46 @@ The following table describes the `CacheMode.CONNECTION` properties:
 
 | Meaning
 
-| `connectionName:<localPort>`
+| connectionName:<localPort>
 
 | The name of the connection generated by the `ConnectionNameStrategy`.
 
-| `openConnections`
+| openConnections
 
 | The number of connection objects representing connections to brokers.
 
-| `channelCacheSize`
+| channelCacheSize
 
 | The currently configured maximum channels that are allowed to be idle.
 
-| `connectionCacheSize`
+| connectionCacheSize
 
 | The currently configured maximum connections that are allowed to be idle.
 
-| `idleConnections`
+| idleConnections
 
 | The number of connections that are currently idle.
 
-| `idleConnectionsHighWater`
+| idleConnectionsHighWater
 
 | The maximum number of connections that have been concurrently idle.
 
-| `idleChannelsTx:<localPort>`
+| idleChannelsTx:<localPort>
 
 | The number of transactional channels that are currently idle (cached) for this connection.
 You can use the `localPort` part of the property name to correlate with connections and channels on the RabbitMQ Admin UI.
 
-| `idleChannelsNotTx:<localPort>`
+| idleChannelsNotTx:<localPort>
 
 | The number of non-transactional channels that are currently idle (cached) for this connection.
 The `localPort` part of the property name can be used to correlate with connections and channels on the RabbitMQ Admin UI.
 
-| `idleChannelsTxHighWater:<localPort>`
+| idleChannelsTxHighWater:<localPort>
 
 | The maximum number of transactional channels that have been concurrently idle (cached).
 The localPort part of the property name can be used to correlate with connections and channels on the RabbitMQ Admin UI.
 
-| `idleChannelsNotTxHighWater:<localPort>`
+| idleChannelsNotTxHighWater:<localPort>
 
 | The maximum number of non-transactional channels have been concurrently idle (cached).
 You can use the `localPort` part of the property name to correlate with connections and channels on the RabbitMQ Admin UI.
@@ -1468,6 +1468,7 @@ This is communicated to the receiving system by setting the `springBatchFormat`
 IMPORTANT: Batched messages are automatically de-batched by listener containers (by using the `springBatchFormat` message header).
 Rejecting any message from a batch causes the entire batch to be rejected.
 
+However, see <<receiving-batch>> for more information.
 
 [[receiving-messages]]
 ==== Receiving Messages
@@ -2725,6 +2726,41 @@ Starting with version 1.5, you can now assign a `group` to the container on the
 This provides a mechanism to get a reference to a subset of containers.
 Adding a `group` attribute causes a bean of type `Collection<MessageListenerContainer>` to be registered with the context with the group name.
 
+[[receiving-batch]]
+===== @RabbitListener with Batching
+
+When receiving a <<template-batching, a batch>> of messages, the de-batching is normally performed by the container and the listener is invoked with one message at at time.
+Starting with version 2.2, you can configure the listener container factory and listener to receive the entire batch in one call, simply set the factory's `batchListener` property, and make the method payload parameter a `List`:
+
+====
+[source, java]
+----
+@Bean
+public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
+    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
+    factory.setConnectionFactory(connectionFactory());
+    factory.setBatchListener(true);
+    return factory;
+}
+
+@RabbitListener(queues = "batch.1")
+public void listen1(List<Thing> in) {
+    ...
+}
+
+// or
+
+@RabbitListener(queues = "batch.2")
+public void listen2(List<Message<Thing>> in) {
+    ...
+}
+----
+====
+
+Setting the `batchListener` property to true automatically turns off the `debatchingEnabled` container property in containers that the factory creates - effectively, the debatching is moved from the container to the listener adapter and the adapter creates the list that is passed to the listener.
+
+A batch-enabled factory cannot be used with a <<annotation-method-selection, multi-method listener>>.
+
 [[using-container-factories]]
 ===== Using Container Factories
 
@@ -4817,7 +4853,7 @@ an aggregate of all containers so designated.
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `channelTransacted`
+| channelTransacted
 (channel-transacted)
 
 | Boolean flag to signal that all messages should be acknowledged in a transaction (either manually or automatically).
@@ -4825,7 +4861,7 @@ a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `acknowledgeMode`
+| acknowledgeMode
 (acknowledge)
 
 a|
@@ -4840,7 +4876,7 @@ See also `txSize`.
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `transactionManager`
+| transactionManager
 (transaction-manager)
 
 | External transaction manager for the operation of the listener.
@@ -4849,7 +4885,7 @@ Also complementary to `channelTransacted` -- if the `Channel` is transacted, its
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `prefetchCount`
+| prefetchCount
 (prefetch)
 
 a| The number of unacknowledged messages that can be outstanding at each consumer.
@@ -4869,7 +4905,7 @@ Also, with low-volume messaging and multiple consumers (including concurrency wi
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `shutdownTimeout`
+| shutdownTimeout
 (N/A)
 
 | When a container shuts down (for example,
@@ -4879,7 +4915,7 @@ Defaults to five seconds.
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `forceCloseChannel`
+| forceCloseChannel
 (N/A)
 
 | If the consumers do not respond to a shutdown within `shutdownTimeout`, if this is `true`, the channel will be closed, causing any unacked messages to be requeued.
@@ -4889,7 +4925,7 @@ You can set it to `false` to revert to the previous behavior.
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `txSize`
+| txSize
 (transaction-size)
 
 | When used with `acknowledgeMode` set to `AUTO`, the container tries to process up to this number of messages before sending an ack (waiting for each one up to the receive timeout setting).
@@ -4899,7 +4935,7 @@ If the `prefetchCount` is less than the `txSize`, it is increased to match the `
 a| image::images/tickmark.png[]
 a|
 
-| `messagesPerAck`
+| messagesPerAck
 (N/A)
 
 | The number of messages to receive between acks.
@@ -4914,7 +4950,7 @@ See also `ackTimeout` in this table.
 a|
 a| image::images/tickmark.png[]
 
-| `ackTimeout`
+| ackTimeout
 (N/A)
 
 | When `messagesPerAck` is set, this timeout is used as an alternative to send an ack.
@@ -4926,7 +4962,7 @@ See also `messagesPerAck` and `monitorInterval` in this table.
 a|
 a| image::images/tickmark.png[]
 
-| `receiveTimeout`
+| receiveTimeout
 (receive-timeout)
 
 | The maximum time to wait for each message.
@@ -4936,7 +4972,7 @@ It has the biggest effect for a transactional `Channel` with `txSize > 1`, since
 a| image::images/tickmark.png[]
 a|
 
-| `autoStartup`
+| autoStartup
 (auto-startup)
 
 | Flag to indicate that the container should start when the `ApplicationContext` does (as part of the `SmartLifecycle` callbacks, which happen after all beans are initialized).
@@ -4945,7 +4981,7 @@ Defaults to `true`, but you can set it to `false` if your broker might not be av
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `phase`
+| phase
 (phase)
 
 | When `autoStartup` is `true`, the lifecycle phase within which this container should start and stop.
@@ -4955,7 +4991,7 @@ The default is `Integer.MAX_VALUE`, meaning the container starts as late as poss
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `adviceChain`
+| adviceChain
 (advice-chain)
 
 | An array of AOP Advice to apply to the listener execution.
@@ -4965,7 +5001,7 @@ Note that simple re-connection after an AMQP error is handled by the `CachingCon
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `taskExecutor`
+| taskExecutor
 (task-executor)
 
 | A reference to a Spring `TaskExecutor` (or standard JDK 1.5+ `Executor`) for executing listener invokers.
@@ -4974,7 +5010,7 @@ Default is a `SimpleAsyncTaskExecutor`, using internally managed threads.
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `errorHandler`
+| errorHandler
 (error-handler)
 
 | A reference to an `ErrorHandler` strategy for handling any uncaught exceptions that may occur during the execution of the MessageListener.
@@ -4983,7 +5019,7 @@ Default: `ConditionalRejectingErrorHandler`
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `consumersPerQueue`
+| consumersPerQueue
 (consumers-per-queue)
 
 | The number of consumers to create for each configured queue.
@@ -4992,7 +5028,7 @@ See <<listener-concurrency>>.
 a|
 a| image::images/tickmark.png[]
 
-| `concurrentConsumers`
+| concurrentConsumers
 (concurrency)
 
 | The number of concurrent consumers to initially start for each listener.
@@ -5001,7 +5037,7 @@ See <<listener-concurrency>>.
 a| image::images/tickmark.png[]
 a|
 
-| `maxConcurrentConsumers`
+| maxConcurrentConsumers
 (max-concurrency)
 
 | The maximum number of concurrent consumers to start, if needed, on demand.
@@ -5011,7 +5047,7 @@ See <<listener-concurrency>>.
 a| image::images/tickmark.png[]
 a|
 
-| `concurrency`
+| concurrency
 (N/A)
 
 | `m-n` The range of concurrent consumers for each listener (min, max).
@@ -5021,7 +5057,7 @@ See <<listener-concurrency>>.
 a| image::images/tickmark.png[]
 a|
 
-| `consumerStartTimeout`
+| consumerStartTimeout
 (N/A)
 
 | The time in milliseconds to wait for a consumer thread to start.
@@ -5034,7 +5070,7 @@ Default: 60000 (one minute).
 a| image::images/tickmark.png[]
 a|
 
-| `startConsumerMinInterval`
+| startConsumerMinInterval
 (min-start-interval)
 
 | The time in milliseconds that must elapse before each new consumer is started on demand.
@@ -5044,7 +5080,7 @@ Default: 10000 (10 seconds).
 a| image::images/tickmark.png[]
 a|
 
-| `stopConsumerMinInterval`
+| stopConsumerMinInterval
 (min-stop-interval)
 
 | The time in milliseconds that must elapse before a consumer is stopped since the last consumer was stopped when an idle consumer is detected.
@@ -5054,7 +5090,7 @@ Default: 60000 (one minute).
 a| image::images/tickmark.png[]
 a|
 
-| `consecutiveActiveTrigger`
+| consecutiveActiveTrigger
 (min-consecutive-active)
 
 | The minimum number of consecutive messages received by a consumer, without a receive timeout occurring, when considering starting a new consumer.
@@ -5065,7 +5101,7 @@ Default: 10.
 a| image::images/tickmark.png[]
 a|
 
-| `consecutiveIdleTrigger`
+| consecutiveIdleTrigger
 (min-consecutive-idle)
 
 | The minimum number of receive timeouts a consumer must experience before considering stopping a consumer.
@@ -5076,7 +5112,7 @@ Default: 10.
 a| image::images/tickmark.png[]
 a|
 
-| `connectionFactory`
+| connectionFactory
 (connection-factory)
 
 | A reference to the `ConnectionFactory`.
@@ -5085,7 +5121,7 @@ When configuring byusing the XML namespace, the default referenced bean name is
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `defaultRequeueRejected`
+| defaultRequeueRejected
 (requeue-rejected)
 
 | Determines whether messages that are rejected because the listener threw an exception should be requeued or not.
@@ -5094,7 +5130,7 @@ Default: `true`.
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `recoveryInterval`
+| recoveryInterval
 (recovery-interval)
 
 | Determines the time in milliseconds between attempts to start a consumer if it fails to start for non-fatal reasons.
@@ -5104,7 +5140,7 @@ Mutually exclusive with `recoveryBackOff`.
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `recoveryBackOff`
+| recoveryBackOff
 (recovery-back-off)
 
 | Specifies the `BackOff` for intervals between attempts to start a consumer if it fails to start for non-fatal reasons.
@@ -5114,7 +5150,7 @@ Mutually exclusive with `recoveryInterval`.
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `exclusive`
+| exclusive
 (exclusive)
 
 | Determines whether the single consumer in this container has exclusive access to the queues.
@@ -5127,7 +5163,7 @@ Default: `false`.
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `rabbitAdmin`
+| rabbitAdmin
 (admin)
 
 | When a listener container listens to at least one auto-delete queue and it is found to be missing during startup, the container uses a `RabbitAdmin` to declare the queue and any related bindings and exchanges.
@@ -5140,7 +5176,7 @@ Defaults to a `RabbitAdmin` that declares all non-conditional elements.
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `missingQueuesFatal`
+| missingQueuesFatal
 (missing-queues-fatal)
 
 a| When set to `true` (default), if none of the configured queues are available on the broker, it is considered fatal.
@@ -5213,7 +5249,7 @@ The default retry properties (3 retries at 5 second intervals) can be overridden
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `mismatchedQueuesFatal`
+| mismatchedQueuesFatal
 (mismatched-queues-fatal)
 
 a| When the container starts, if this property is `true` (default: `false`), the container checks that all queues declared in the context are compatible with queues already on the broker.
@@ -5237,7 +5273,7 @@ Applications using lazy listener beans should check the queue arguments before g
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `autoDeclare`
+| autoDeclare
 (auto-declare)
 
 a| When set to `true` (default), the container uses a `RabbitAdmin` to redeclare all AMQP objects (queues, exchanges, bindings), if it detects that at least one of its queues is missing during startup, perhaps because it is an `auto-delete` or an expired queue, but the redeclaration proceeds if the queue is missing for any reason.
@@ -5252,7 +5288,7 @@ Starting with version 1.6, for `autoDeclare` to work, there must be exactly one
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `declarationRetries`
+| declarationRetries
 (declaration-retries)
 
 | The number of retry attempts when passive queue declaration fails.
@@ -5263,7 +5299,7 @@ Default: Three retries (for a total of four attempts).
 a| image::images/tickmark.png[]
 a|
 
-| `failedDeclarationRetryInterval`
+| failedDeclarationRetryInterval
 (failed-declaration-retry-
 interval)
 
@@ -5274,7 +5310,7 @@ Default: 5000 (five seconds).
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `retryDeclarationInterval`
+| retryDeclarationInterval
 (missing-queue-retry-interval)
 
 | If a subset of the configured queues are available during consumer initialization, the consumer starts consuming from those queues.
@@ -5287,7 +5323,7 @@ Default: 60000 (one minute).
 a| image::images/tickmark.png[]
 a|
 
-| `consumerTagStrategy`
+| consumerTagStrategy
 (consumer-tag-strategy)
 
 | Set an implementation of <<consumerTags, ConsumerTagStrategy>>, enabling the creation of a (unique) tag for each consumer.
@@ -5295,7 +5331,7 @@ a|
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `idleEventInterval`
+| idleEventInterval
 (idle-event-interval)
 
 | See <<idle-containers>>.
@@ -5303,7 +5339,7 @@ a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `monitorInterval`
+| monitorInterval
 (monitor-interval)
 
 | With the DMLC, a task is scheduled to run at this interval to monitor the state of the consumers and recover any that have failed.
@@ -5311,7 +5347,7 @@ a| image::images/tickmark.png[]
 a|
 a| image::images/tickmark.png[]
 
-| `taskScheduler`
+| taskScheduler
 (task-scheduler)
 
 | With the DMLC, the scheduler used to run the monitor task at the 'monitorInterval'.
@@ -5319,7 +5355,7 @@ a| image::images/tickmark.png[]
 a|
 a| image::images/tickmark.png[]
 
-| `exclusiveConsumer`
+| exclusiveConsumer
 `ExceptionLogger`
 (N/A)
 
@@ -5329,7 +5365,7 @@ By default, this is logged at the `WARN` level.
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `statefulRetryFatalWithNullMessageId`
+| statefulRetryFatalWithNullMessageId
 (N/A)
 
 | When using a stateful retry advice, if a message with a missing `messageId` property is received, it is considered
@@ -5339,7 +5375,7 @@ Set this to `false` to discard (or route to a dead-letter queue) such messages.
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `alwaysRequeueWithTxManagerRollback`
+| alwaysRequeueWithTxManagerRollback
 (N/A)
 
 | Set to `true` to always requeue messages on rollback when a transaction manager is configured.
@@ -5347,7 +5383,7 @@ a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `noLocal`
+| noLocal
 (N/A)
 
 | Set to `true` to disable delivery from the server to consumers messages published on the same channel's connection.
@@ -5355,7 +5391,7 @@ a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
-| `afterReceivePostProcessors`
+| afterReceivePostProcessors
 (N/A)
 
 | An array of `MessagePostProcessor` instances that are invoked before invoking the listener.
@@ -5366,6 +5402,26 @@ If a post processor returns `null`, the message is discarded (and acknowledged,
 a| image::images/tickmark.png[]
 a| image::images/tickmark.png[]
 
+| debatchingEnabled
+(N/A)
+
+| When true, the listener container will debatch batched messages and invoke the listener with each message from the batch.
+Default true.
+See <<template-batching>> and <<receiving-batch>>.
+
+a| image::images/tickmark.png[]
+a| image::images/tickmark.png[]
+
+| batchingStrategy
+(N/A)
+
+| The strategy used when debatchng messages.
+Default `SimpleDebatchingStrategy`.
+See <<template-batching>> and <<receiving-batch>>.
+
+a| image::images/tickmark.png[]
+a| image::images/tickmark.png[]
+
 |===
 
 [[listener-concurrency]]

src/reference/asciidoc/whats-new.adoc

@@ -10,6 +10,8 @@ This section describes the changes between version 2.1 and version 2.2.
 You can now configure an `executor` on each listener, overriding the factory configuration, to more easily identify threads associated with the listener.
 See <<async-annotation-driven-enable>> for more information.
 
+When using <<receiving-batch,batching>>, `@RabbitListener` methods can now receive a complete batch of messages in one call instead of getting them one at at time.
+
 ===== AMQP Logging Appenders Changes
 
 The Log4J and Logback `AmqpAppender` s now support a `verifyHostname` SSL option.