diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc index 083ef3520..a9b060506 100644 --- a/docs/src/main/asciidoc/overview.adoc +++ b/docs/src/main/asciidoc/overview.adoc @@ -424,9 +424,188 @@ This allows you to add arguments that are not currently directly supported by th [[rabbit-receiving-batch]] === Receiving Batched Messages +With the RabbitMQ binder, there are two types of batches handled by consumer bindings: + +==== Batches Created by Producers + Normally, if a producer binding has `batch-enabled=true` (see <>), or a message is created by a `BatchingRabbitTemplate`, elements of the batch are returned as individual calls to the listener method. Starting with version 3.0, any such batch can be presented as a `List` to the listener method if `spring.cloud.stream.bindings..consumer.batch-mode` is set to `true`. +==== Consumer-side Batching + +Starting with version 3.1, the consumer can be configured to assemble multiple inbound messages into a batch which is presented to the application as a `List` of converted payloads. +The following simple application demonstrates how to use this technique: + +==== +[source, properties] +---- +spring.cloud.stream.bindings.input-in-0.group=someGroup + +spring.cloud.stream.bindings.input-in-0.consumer.batch-mode=true + +spring.cloud.stream.rabbit.bindings.input-in-0.consumer.enable-batching=true +spring.cloud.stream.rabbit.bindings.input-in-0.consumer.batch-size=10 +spring.cloud.stream.rabbit.bindings.input-in-0.consumer.receive-timeout=200 +---- +==== + +==== +[source, java] +---- +@SpringBootApplication +public class Application { + + public static void main(String[] args) { + SpringApplication.run(Application.class, args); + } + + @Bean + Consumer> input() { + return list -> { + System.out.println("Received " + list.size()); + list.forEach(thing -> { + System.out.println(thing); + + // ... + + }); + }; + } + + @Bean + public ApplicationRunner runner(RabbitTemplate template) { + return args -> { + template.convertAndSend("input-in-0.someGroup", "{\"field\":\"value1\"}"); + template.convertAndSend("input-in-0.someGroup", "{\"field\":\"value2\"}"); + }; + } + + public static class Thing { + + private String field; + + public Thing() { + } + + public Thing(String field) { + this.field = field; + } + + public String getField() { + return this.field; + } + + public void setField(String field) { + this.field = field; + } + + @Override + public String toString() { + return "Thing [field=" + this.field + "]"; + } + + } + +} +---- +==== + +==== +[source] +---- +Received 2 +Thing [field=value1] +Thing [field=value2] +---- +==== + +The number of messages in a batch is specified by the `batch-size` and `receive-timeout` properties; if the `receive-timeout` elapses with no new messages, a "short" batch is delivered. + +IMPORTANT: Consumer-side batching is only supported with `container-type=simple` (the default). + +If you wish to examine headers of consumer-side batched messages, you should consume `Message>`; the headers are a `List>` in a header `AmqpInboundChannelAdapter.CONSOLIDATED_HEADERS`, with the headers for each payload element in the corresponding index. +Again, here is a simple example: + +==== +[source, java] +---- +@SpringBootApplication +public class Application { + + public static void main(String[] args) { + SpringApplication.run(Application.class, args); + } + + @Bean + Consumer>> input() { + return msg -> { + List things = msg.getPayload(); + System.out.println("Received " + things.size()); + @SuppressWarnings("unchecked") + List> headers = + (List>) msg.getHeaders().get(AmqpInboundChannelAdapter.CONSOLIDATED_HEADERS); + for (int i = 0; i < things.size(); i++) { + System.out.println(things.get(i) + " myHeader=" + headers.get(i).get("myHeader")); + + // ... + + } + }; + } + + @Bean + public ApplicationRunner runner(RabbitTemplate template) { + return args -> { + template.convertAndSend("input-in-0.someGroup", "{\"field\":\"value1\"}", msg -> { + msg.getMessageProperties().setHeader("myHeader", "headerValue1"); + return msg; + }); + template.convertAndSend("input-in-0.someGroup", "{\"field\":\"value2\"}", msg -> { + msg.getMessageProperties().setHeader("myHeader", "headerValue2"); + return msg; + }); + }; + } + + public static class Thing { + + private String field; + + public Thing() { + } + + public Thing(String field) { + this.field = field; + } + + public String getfield() { + return this.field; + } + + public void setfield(String field) { + this.field = field; + } + + @Override + public String toString() { + return "Thing [field=" + this.field + "]"; + } + + } + +} +---- +==== + +==== +[source] +---- +Received 2 +Thing [field=value1] myHeader=headerValue1 +Thing [field=value2] myHeader=headerValue2 +---- +==== + [[rabbit-prod-props]] === Rabbit Producer Properties diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java index 916488d81..720647d5a 100644 --- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java @@ -125,6 +125,21 @@ public class RabbitConsumerProperties extends RabbitCommonProperties { */ private String anonymousGroupPrefix = "anonymous."; + /** + * When true, the listener container will assemble a list from multiple messages, + * according to the batchSize and receiveTimeout properties. Only applies with + * {@link ContainerType#SIMPLE}. + */ + private boolean enableBatching; + + /** + * How long to block waiting to receive messages; increasing from the default 1 second + * will make the binding less responsive to stop requests. When enableConsumerBatching + * is true, a short batch may be emitted if this time elapses before the batchSize is + * satisfied. Only applies with {@link ContainerType#SIMPLE}. + */ + private Long receiveTimeout; + public boolean isTransacted() { return transacted; } @@ -317,4 +332,20 @@ public class RabbitConsumerProperties extends RabbitCommonProperties { this.anonymousGroupPrefix = anonymousGroupPrefix; } + public boolean isEnableBatching() { + return this.enableBatching; + } + + public void setEnableBatching(boolean enableBatching) { + this.enableBatching = enableBatching; + } + + public Long getReceiveTimeout() { + return this.receiveTimeout; + } + + public void setReceiveTimeout(Long receiveTimeout) { + this.receiveTimeout = receiveTimeout; + } + } diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index 76b11a7d5..d5974aea7 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -88,6 +88,7 @@ import org.springframework.integration.StaticMessageHeaderAccessor; import org.springframework.integration.acks.AcknowledgmentCallback; import org.springframework.integration.acks.AcknowledgmentCallback.Status; import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter; +import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter.BatchMode; import org.springframework.integration.amqp.inbound.AmqpMessageSource; import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint; import org.springframework.integration.amqp.support.AmqpMessageHeaderErrorMessageStrategy; @@ -446,16 +447,17 @@ public class RabbitMessageChannelBinder extends Assert.state(!HeaderMode.embeddedHeaders.equals(properties.getHeaderMode()), "the RabbitMQ binder does not support embedded headers since RabbitMQ supports headers natively"); String destination = consumerDestination.getName(); - boolean directContainer = properties.getExtension().getContainerType() + RabbitConsumerProperties extension = properties.getExtension(); + boolean directContainer = extension.getContainerType() .equals(ContainerType.DIRECT); AbstractMessageListenerContainer listenerContainer = directContainer ? new DirectMessageListenerContainer(this.connectionFactory) : new SimpleMessageListenerContainer(this.connectionFactory); listenerContainer - .setAcknowledgeMode(properties.getExtension().getAcknowledgeMode()); - listenerContainer.setChannelTransacted(properties.getExtension().isTransacted()); + .setAcknowledgeMode(extension.getAcknowledgeMode()); + listenerContainer.setChannelTransacted(extension.isTransacted()); listenerContainer - .setDefaultRequeueRejected(properties.getExtension().isRequeueRejected()); + .setDefaultRequeueRejected(extension.isRequeueRejected()); int concurrency = properties.getConcurrency(); concurrency = concurrency > 0 ? concurrency : 1; if (directContainer) { @@ -466,9 +468,9 @@ public class RabbitMessageChannelBinder extends setSMLCProperties(properties, (SimpleMessageListenerContainer) listenerContainer, concurrency); } - listenerContainer.setPrefetchCount(properties.getExtension().getPrefetch()); + listenerContainer.setPrefetchCount(extension.getPrefetch()); listenerContainer - .setRecoveryInterval(properties.getExtension().getRecoveryInterval()); + .setRecoveryInterval(extension.getRecoveryInterval()); listenerContainer.setTaskExecutor( new SimpleAsyncTaskExecutor(consumerDestination.getName() + "-")); String[] queues = StringUtils.tokenizeToStringArray(destination, ",", true, true); @@ -476,12 +478,12 @@ public class RabbitMessageChannelBinder extends listenerContainer.setAfterReceivePostProcessors(this.decompressingPostProcessor); listenerContainer.setMessagePropertiesConverter( RabbitMessageChannelBinder.inboundMessagePropertiesConverter); - listenerContainer.setExclusive(properties.getExtension().isExclusive()); + listenerContainer.setExclusive(extension.isExclusive()); listenerContainer - .setMissingQueuesFatal(properties.getExtension().getMissingQueuesFatal()); - if (properties.getExtension().getFailedDeclarationRetryInterval() != null) { + .setMissingQueuesFatal(extension.getMissingQueuesFatal()); + if (extension.getFailedDeclarationRetryInterval() != null) { listenerContainer.setFailedDeclarationRetryInterval( - properties.getExtension().getFailedDeclarationRetryInterval()); + extension.getFailedDeclarationRetryInterval()); } if (getApplicationEventPublisher() != null) { listenerContainer @@ -492,10 +494,10 @@ public class RabbitMessageChannelBinder extends } getContainerCustomizer().configure(listenerContainer, consumerDestination.getName(), group); - if (StringUtils.hasText(properties.getExtension().getConsumerTagPrefix())) { + if (StringUtils.hasText(extension.getConsumerTagPrefix())) { final AtomicInteger index = new AtomicInteger(); listenerContainer.setConsumerTagStrategy( - q -> properties.getExtension().getConsumerTagPrefix() + "#" + q -> extension.getConsumerTagPrefix() + "#" + index.getAndIncrement()); } listenerContainer.afterPropertiesSet(); @@ -506,7 +508,7 @@ public class RabbitMessageChannelBinder extends adapter.setBeanFactory(this.getBeanFactory()); adapter.setBeanName("inbound." + destination); DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.inboundMapper(); - mapper.setRequestHeaderNames(properties.getExtension().getHeaderPatterns()); + mapper.setRequestHeaderNames(extension.getHeaderPatterns()); adapter.setHeaderMapper(mapper); ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure( consumerDestination, group, properties); @@ -519,6 +521,10 @@ public class RabbitMessageChannelBinder extends adapter.setErrorChannel(errorInfrastructure.getErrorChannel()); } adapter.setMessageConverter(passThoughConverter); + if (properties.isBatchMode() && extension.isEnableBatching() + && ContainerType.SIMPLE.equals(extension.getContainerType())) { + adapter.setBatchMode(BatchMode.EXTRACT_PAYLOADS_WITH_HEADERS); + } return adapter; } @@ -526,16 +532,24 @@ public class RabbitMessageChannelBinder extends ExtendedConsumerProperties properties, SimpleMessageListenerContainer listenerContainer, int concurrency) { + RabbitConsumerProperties extension = properties.getExtension(); listenerContainer.setConcurrentConsumers(concurrency); - int maxConcurrency = properties.getExtension().getMaxConcurrency(); + int maxConcurrency = extension.getMaxConcurrency(); if (maxConcurrency > concurrency) { listenerContainer.setMaxConcurrentConsumers(maxConcurrency); } listenerContainer.setDeBatchingEnabled(!properties.isBatchMode()); - listenerContainer.setBatchSize(properties.getExtension().getBatchSize()); - if (properties.getExtension().getQueueDeclarationRetries() != null) { + listenerContainer.setBatchSize(extension.getBatchSize()); + if (extension.getQueueDeclarationRetries() != null) { listenerContainer.setDeclarationRetries( - properties.getExtension().getQueueDeclarationRetries()); + extension.getQueueDeclarationRetries()); + } + if (properties.isBatchMode() && extension.isEnableBatching()) { + listenerContainer.setConsumerBatchEnabled(true); + listenerContainer.setDeBatchingEnabled(true); + } + if (extension.getReceiveTimeout() != null) { + listenerContainer.setReceiveTimeout(extension.getReceiveTimeout()); } } diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index 26cdab37d..8ec89f453 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java @@ -1557,7 +1557,7 @@ public class RabbitBinderTests extends @SuppressWarnings("unchecked") @Test - public void testConsumerBatching() throws Exception { + public void testProducerBatching() throws Exception { RabbitTestBinder binder = getBinder(); ExtendedProducerProperties producerProperties = createProducerProperties(); producerProperties.getExtension() @@ -1570,7 +1570,44 @@ public class RabbitBinderTests extends DirectChannel output = createBindableChannel("output", createProducerBindingProperties(producerProperties)); - output.setBeanName("consumerBatchingProducer"); + output.setBeanName("producerBatchingProducer"); + Binding producerBinding = binder.bindProducer("p.batching.0", + output, producerProperties); + + QueueChannel input = new QueueChannel(); + input.setBeanName("producerBatchingConsumer"); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.setBatchMode(true); + consumerProperties.getExtension().setBatchSize(2); + Binding consumerBinding = binder.bindConsumer("p.batching.0", + "producerBatching", input, consumerProperties); + output.send(new GenericMessage<>("foo".getBytes())); + output.send(new GenericMessage<>("bar".getBytes())); + + + Message in = input.receive(10000); + assertThat(in).isNotNull(); + assertThat(in.getPayload()).isInstanceOf(List.class); + List payload = (List) in.getPayload(); + assertThat(payload).hasSize(2); + assertThat(payload.get(0)).isEqualTo("foo".getBytes()); + assertThat(payload.get(1)).isEqualTo("bar".getBytes()); + + producerBinding.unbind(); + consumerBinding.unbind(); + } + + @SuppressWarnings("unchecked") + @Test + public void testConsumerBatching() throws Exception { + RabbitTestBinder binder = getBinder(); + ExtendedProducerProperties producerProperties = createProducerProperties(); + producerProperties.getExtension() + .setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT); + + DirectChannel output = createBindableChannel("output", + createProducerBindingProperties(producerProperties)); + output.setBeanName("consumerBatching.Producer"); Binding producerBinding = binder.bindProducer("c.batching.0", output, producerProperties); @@ -1579,6 +1616,7 @@ public class RabbitBinderTests extends ExtendedConsumerProperties consumerProperties = createConsumerProperties(); consumerProperties.setBatchMode(true); consumerProperties.getExtension().setBatchSize(2); + consumerProperties.getExtension().setEnableBatching(true); Binding consumerBinding = binder.bindConsumer("c.batching.0", "consumerBatching", input, consumerProperties); output.send(new GenericMessage<>("foo".getBytes()));