diff --git a/README.adoc b/README.adoc index 90c103312..2e72f175d 100644 --- a/README.adoc +++ b/README.adoc @@ -445,9 +445,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 @@ -509,6 +688,7 @@ confirmAckChannel:: When `errorChannelEnabled` is true, a channel to which to send positive delivery acknowledgments (aka publisher confirms). If the channel does not exist, a `DirectChannel` is registered with this name. The connection factory must be configured to enable publisher confirms. +Mutually exclusive with `useConfirmHeader`. + Default: `nullChannel` (acks are discarded). deadLetterQueueName:: @@ -721,10 +901,131 @@ Default time (in milliseconds) to live to apply to the queue when declared. Applies only when `requiredGroups` are provided and then only to those groups. + Default: `no limit` +useConfirmHeader:: +See <>. +Mutually exclusive with `confirmAckChannel`. ++ NOTE: In the case of RabbitMQ, content type headers can be set by external applications. Spring Cloud Stream supports them as part of an extended internal protocol used for any type of transport -- including transports, such as Kafka (prior to 0.11), that do not natively support headers. +[[publisher-confirms]] +=== Publisher Confirms + +There are two mechanisms to get the result of publishing a message; in each case, the connection factory must have `publisherConfirmType` set `ConfirmType.CORRELATED`. +The "legacy" mechanism is to set the `confirmAckChannel` to the bean name of a message channel from which you can retrieve the confirmations asynchronously; negative acks are sent to the error channel (if enabled) - see <>. + +The preferred mechanism, added in version 3.1 is to use a correlation data header and wait for the result via its `Future` property. +This is particularly useful with a batch listener because you can send multiple messages before waiting for the result. +To use this technique, set the `useConfirmHeader` property to true +The following simple application is an example of using 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.source=output +spring.cloud.stream.bindings.output-out-0.producer.error-channel-enabled=true + +spring.cloud.stream.rabbit.bindings.output-out-0.producer.useConfirmHeader=true +spring.cloud.stream.rabbit.bindings.input-in-0.consumer.auto-bind-dlq=true +spring.cloud.stream.rabbit.bindings.input-in-0.consumer.batch-size=10 + +spring.rabbitmq.publisher-confirm-type=correlated +spring.rabbitmq.publisher-returns=true +---- +==== + +==== +[source, java] +---- +@SpringBootApplication +public class Application { + + private static final Logger log = LoggerFactory.getLogger(Application.class); + + public static void main(String[] args) { + SpringApplication.run(Application.class, args); + } + + @Autowired + private StreamBridge bridge; + + @Bean + Consumer> input() { + return list -> { + List results = new ArrayList<>(); + list.forEach(str -> { + log.info("Received: " + str); + MyCorrelationData corr = new MyCorrelationData(UUID.randomUUID().toString(), str); + results.add(corr); + this.bridge.send("output-out-0", MessageBuilder.withPayload(str.toUpperCase()) + .setHeader(AmqpHeaders.PUBLISH_CONFIRM_CORRELATION, corr) + .build()); + }); + results.forEach(correlation -> { + try { + Confirm confirm = correlation.getFuture().get(10, TimeUnit.SECONDS); + log.info(confirm + " for " + correlation.getPayload()); + if (correlation.getReturnedMessage() != null) { + log.error("Message for " + correlation.getPayload() + " was returned "); + + // try to re-publish, send a DLQ, etc + + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + e.printStackTrace(); + } + catch (ExecutionException | TimeoutException e) { + e.printStackTrace(); + } + }); + }; + } + + + @Bean + public ApplicationRunner runner(BatchingRabbitTemplate template) { + return args -> IntStream.range(0, 10).forEach(i -> + template.convertAndSend("input-in-0", "input-in-0.rbgh303", "foo" + i)); + } + + @Bean + public BatchingRabbitTemplate template(CachingConnectionFactory cf, TaskScheduler taskScheduler) { + BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(10, 1000000, 1000); + return new BatchingRabbitTemplate(cf, batchingStrategy, taskScheduler); + } + +} + +class MyCorrelationData extends CorrelationData { + + private final String payload; + + MyCorrelationData(String id, String payload) { + super(id); + this.payload = payload; + } + + public String getPayload() { + return this.payload; + } + +} +---- +==== + +As you can see, we send each message and then await for the publication results. +If the messages can't be routed, then correlation data is populated with the returned message before the future is completed. + +IMPORTANT: The correlation data must be provided with a unique `id` so that the framework can perform the correlation. + +You cannot set both `useConfirmHeader` and `confirmAckChannel` but you can still receive returned messages in the error channel when `useConfirmHeader` is true, but using the correlation header is more convenient. + == Using Existing Queues/Exchanges By default, the binder will automatically provision a topic exchange with the name being derived from the value of the destination binding property ``. @@ -835,6 +1136,7 @@ RabbitMQ has two types of send failures: The latter is rare. According to the RabbitMQ documentation "[A nack] will only be delivered if an internal error occurs in the Erlang process responsible for a queue.". +You can also get a negative acknowledgment if you publish to a bounded queue with `reject-publish` queue overflow behavior. As well as enabling producer error channels (as described in "`<>`"), the RabbitMQ binder only sends messages to the channels if the connection factory is appropriately configured, as follows: @@ -855,6 +1157,8 @@ The payload of the `ErrorMessage` for a returned message is a `ReturnedAmqpMessa * `exchange`: The exchange to which the message was published. * `routingKey`: The routing key used when the message was published. +Also see <> for an alternative mechanism to receive returned messages. + For negatively acknowledged confirmations, the payload is a `NackedAmqpMessageException` with the following properties: * `failedMessage`: The spring-messaging `Message` that failed to be sent. diff --git a/docs/src/main/asciidoc/_configprops.adoc b/docs/src/main/asciidoc/_configprops.adoc index 4542a85d5..9f011f67a 100644 --- a/docs/src/main/asciidoc/_configprops.adoc +++ b/docs/src/main/asciidoc/_configprops.adoc @@ -20,6 +20,7 @@ |spring.cloud.stream.metrics.properties | | Application properties that should be added to the metrics payload For example: `spring.application**`. |spring.cloud.stream.metrics.schedule-interval | 60s | Interval expressed as Duration for scheduling metrics snapshots publishing. Defaults to 60 seconds |spring.cloud.stream.override-cloud-connectors | false | This property is only applicable when the cloud profile is active and Spring Cloud Connectors are provided with the application. If the property is false (the default), the binder detects a suitable bound service (for example, a RabbitMQ service bound in Cloud Foundry for the RabbitMQ binder) and uses it for creating connections (usually through Spring Cloud Connectors). When set to true, this property instructs binders to completely ignore the bound services and rely on Spring Boot properties (for example, relying on the spring.rabbitmq.* properties provided in the environment for the RabbitMQ binder). The typical usage of this property is to be nested in a customized environment when connecting to multiple systems. +|spring.cloud.stream.pollable-source | none | A semi-colon delimited list of binding names of pollable sources. Binding names follow the same naming convention as functions. For example, name '...pollable-source=foobar' will be accessible as 'foobar-iin-0'' binding |spring.cloud.stream.poller.cron | | Cron expression value for the Cron Trigger. |spring.cloud.stream.poller.fixed-delay | 1000 | Fixed delay for default poller. |spring.cloud.stream.poller.initial-delay | 0 | Initial delay for periodic triggers. diff --git a/pom.xml b/pom.xml index ab103374e..7647b8c64 100644 --- a/pom.xml +++ b/pom.xml @@ -7,7 +7,7 @@ org.springframework.cloud spring-cloud-build - 3.0.0-SNAPSHOT + 3.0.0-M4