diff --git a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/dlq.adoc b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/dlq.adoc index 1d23197b3..e4754ee9a 100644 --- a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/dlq.adoc +++ b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/dlq.adoc @@ -1,19 +1,20 @@ [[rabbit-dlq-processing]] == Dead-Letter Queue Processing -Because it can't be anticipated how users would want to dispose of dead-lettered messages, the framework does not provide any standard mechanism to handle them. +Because you cannot anticipate how users would want to dispose of dead-lettered messages, the framework does not provide any standard mechanism to handle them. If the reason for the dead-lettering is transient, you may wish to route the messages back to the original queue. However, if the problem is a permanent issue, that could cause an infinite loop. -The following `spring-boot` application is an example of how to route those messages back to the original queue, but moves them to a third "parking lot" queue after three attempts. -The second example utilizes the https://www.rabbitmq.com/blog/2015/04/16/scheduling-messages-with-rabbitmq/[RabbitMQ Delayed Message Exchange] to introduce a delay to the requeued message. +The following Spring Boot application shows an example of how to route those messages back to the original queue but moves them to a third "`parking lot`" queue after three attempts. +The second example uses the https://www.rabbitmq.com/blog/2015/04/16/scheduling-messages-with-rabbitmq/[RabbitMQ Delayed Message Exchange] to introduce a delay to the re-queued message. In this example, the delay increases for each attempt. -These examples use a `@RabbitListener` to receive messages from the DLQ, you could also use `RabbitTemplate.receive()` in a batch process. +These examples use a `@RabbitListener` to receive messages from the DLQ. +You could also use `RabbitTemplate.receive()` in a batch process. The examples assume the original destination is `so8400in` and the consumer group is `so8400`. === Non-Partitioned Destinations -The first two examples are when the destination is **not** partitioned. +The first two examples are for when the destination is *not* partitioned: [source, java] ---- @@ -125,11 +126,11 @@ public class ReRouteDlqApplication { === Partitioned Destinations -With partitioned destinations, there is one DLQ for all partitions and we determine the original queue from the headers. +With partitioned destinations, there is one DLQ for all partitions. We determine the original queue from the headers. -==== republishToDlq=false +==== `republishToDlq=false` -When `republishToDlq` is `false`, RabbitMQ publishes the message to the DLX/DLQ with an `x-death` header containing information about the original destination. +When `republishToDlq` is `false`, RabbitMQ publishes the message to the DLX/DLQ with an `x-death` header containing information about the original destination, as shown in the following example: [source, java] ---- @@ -184,9 +185,9 @@ public class ReRouteDlqApplication { } ---- -==== republishToDlq=true +==== `republishToDlq=true` -When `republishToDlq` is `true`, the republishing recoverer adds the original exchange and routing key to headers. +When `republishToDlq` is `true`, the republishing recoverer adds the original exchange and routing key to headers, as shown in the following example: [source, java] ---- diff --git a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/index.adoc b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/index.adoc index a38c6f858..73e87b6f7 100644 --- a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/index.adoc +++ b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/index.adoc @@ -1,6 +1,6 @@ [[spring-cloud-stream-binder-rabbit-reference]] = Spring Cloud Stream RabbitMQ Binder Reference Guide -Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinathan, Gunnar Hillert, Mark Pollack, Patrick Peralta, Glenn Renfro, Thomas Risberg, Dave Syer, David Turanski, Janne Valkealahti, Benjamin Klein, Gary Russell +Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinathan, Gunnar Hillert, Mark Pollack, Patrick Peralta, Glenn Renfro, Thomas Risberg, Dave Syer, David Turanski, Janne Valkealahti, Benjamin Klein, Gary Russell, Jay Bryant :doctype: book :toc: :toclevels: 4 diff --git a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc index dd5ffc377..31d192d09 100644 --- a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc +++ b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc @@ -6,7 +6,7 @@ It contains information about its design, usage and configuration options, as we == Usage -For using the RabbitMQ binder, you just need to add it to your Spring Cloud Stream application, using the following Maven coordinates: +To use the RabbitMQ binder, you can add it to your Spring Cloud Stream application, by using the following Maven coordinates: [source,xml] ---- @@ -16,7 +16,7 @@ For using the RabbitMQ binder, you just need to add it to your Spring Cloud Stre ---- -Alternatively, you can also use the Spring Cloud Stream RabbitMQ Starter. +Alternatively, you can use the Spring Cloud Stream RabbitMQ Starter, as follows: [source,xml] ---- @@ -28,268 +28,276 @@ Alternatively, you can also use the Spring Cloud Stream RabbitMQ Starter. == RabbitMQ Binder Overview -A simplified diagram of how the RabbitMQ binder operates can be seen below. +The following simplified diagram shows how the RabbitMQ binder operates: .RabbitMQ Binder image::rabbit-binder.png[width=300,scaledwidth="50%"] -The RabbitMQ Binder implementation maps each destination to a `TopicExchange` (by default). -For each consumer group, a `Queue` will be bound to that `TopicExchange`. -Each consumer instance have a corresponding RabbitMQ `Consumer` instance for its group's `Queue`. -For partitioned producers/consumers the queues are suffixed with the partition index and use the partition index as routing key. -For anonymous consumers (no `group` property) an auto-delete queue is used, with a randomized unique name. +By default, the RabbitMQ Binder implementation maps each destination to a `TopicExchange`. +For each consumer group, a `Queue` is bound to that `TopicExchange`. +Each consumer instance has a corresponding RabbitMQ `Consumer` instance for its group's `Queue`. +For partitioned producers and consumers, the queues are suffixed with the partition index and use the partition index as the routing key. +For anonymous consumers (those with no `group` property), an auto-delete queue (with a randomized unique name) is used. -Using the optional `autoBindDlq` option, you can configure the binder to create and configure dead-letter queues (DLQs) (and a dead-letter exchange `DLX` as well as routing infrastructure). +By using the optional `autoBindDlq` option, you can configure the binder to create and configure dead-letter queues (DLQs) (and a dead-letter exchange `DLX`, as well as routing infrastructure). By default, the dead letter queue has the name of the destination, appended with `.dlq`. -If retry is enabled (`maxAttempts > 1`) failed messages will be delivered to the DLQ after retries are exhausted. -If retry is disabled (`maxAttempts = 1`), you should set `requeueRejected` to `false` (default) so that a failed message will be routed to the DLQ, instead of being requeued. -In addition, `republishToDlq` causes the binder to publish a failed message to the DLQ (instead of rejecting it); this enables additional information to be added to the message in headers, such as the stack trace in the `x-exception-stacktrace` header. -This option does not need retry enabled; you can republish a failed message after just one attempt. -Starting with _version 1.2_, you can configure the delivery mode of republished messages; see property `republishDeliveryMode`. +If retry is enabled (`maxAttempts > 1`), failed messages are delivered to the DLQ after retries are exhausted. +If retry is disabled (`maxAttempts = 1`), you should set `requeueRejected` to `false` (the default) so that failed messages are routed to the DLQ, instead of being re-queued. +In addition, `republishToDlq` causes the binder to publish a failed message to the DLQ (instead of rejecting it). +This feature lets additional information (such as the stack trace in the `x-exception-stacktrace` header) be added to the message in headers. +This option does not need retry enabled. +You can republish a failed message after just one attempt. +Starting with version 1.2, you can configure the delivery mode of republished messages. +See the <>. -IMPORTANT: Setting `requeueRejected` to `true` (with `republishToDlq=false` ) will cause the message to be requeued and redelivered continually, which is likely not what you want unless the reason for the failure is transient. -In general, it's better to enable retry within the binder by setting `maxAttempts` to greater than one, or set `republishToDlq` to `true`. +IMPORTANT: Setting `requeueRejected` to `true` (with `republishToDlq=false` ) causes the message to be re-queued and redelivered continually, which is likely not what you want unless the reason for the failure is transient. +In general, you should enable retry within the binder by setting `maxAttempts` to greater than one or by setting `republishToDlq` to `true`. See <> for more information about these properties. The framework does not provide any standard mechanism to consume dead-letter messages (or to re-route them back to the primary queue). Some options are described in <>. -[NOTE] -==== -When *multiple* RabbitMQ binders are used in a Spring Cloud Stream application, it is important to disable 'RabbitAutoConfiguration' to avoid the same configuration from `RabbitAutoConfiguration` being applied to the two binders. -Exclude the class using the `@SpringBootApplication` annotation. -==== +NOTE: When multiple RabbitMQ binders are used in a Spring Cloud Stream application, it is important to disable 'RabbitAutoConfiguration' to avoid the same configuration from `RabbitAutoConfiguration` being applied to the two binders. +You can exclude the class by using the `@SpringBootApplication` annotation. -Starting with _version 2.0_, the `RabbitMessageChannelBinder` sets the `RabbitTemplate.userPublisherConnection` property to true so that the non-transactional producers will avoid dead locks on consumers which can happen if cached connections are blocked because of https://www.rabbitmq.com/memory.html[Memory Alarm] on Broker. +Starting with version 2.0, the `RabbitMessageChannelBinder` sets the `RabbitTemplate.userPublisherConnection` property to `true` so that the non-transactional producers avoid deadlocks on consumers, which can happen if cached connections are blocked because of a https://www.rabbitmq.com/memory.html[memory alarm] on the broker. == Configuration Options This section contains settings specific to the RabbitMQ Binder and bound channels. -For general binding configuration options and properties, -please refer to the https://github.com/spring-cloud/spring-cloud-stream/blob/master/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc#configuration-options[Spring Cloud Stream core documentation]. +For general binding configuration options and properties, see the https://github.com/spring-cloud/spring-cloud-stream/blob/master/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc#configuration-options[Spring Cloud Stream core documentation]. [[rabbit-binder-properties]] === RabbitMQ Binder Properties -By default, the RabbitMQ binder uses Spring Boot's `ConnectionFactory`, and it therefore supports all Spring Boot configuration options for RabbitMQ. -(For reference, consult the http://docs.spring.io/spring-boot/docs/current/reference/htmlsingle/#common-application-properties[Spring Boot documentation]). +By default, the RabbitMQ binder uses Spring Boot's `ConnectionFactory`. +Conseuqently, it supports all Spring Boot configuration options for RabbitMQ. +(For reference, see the http://docs.spring.io/spring-boot/docs/current/reference/htmlsingle/#common-application-properties[Spring Boot documentation]). RabbitMQ configuration options use the `spring.rabbitmq` prefix. In addition to Spring Boot options, the RabbitMQ binder supports the following properties: spring.cloud.stream.rabbit.binder.adminAddresses:: - A comma-separated list of RabbitMQ management plugin URLs. +A comma-separated list of RabbitMQ management plugin URLs. Only used when `nodes` contains more than one entry. Each entry in this list must have a corresponding entry in `spring.rabbitmq.addresses`. -**Only needed if you are using a RabbitMQ cluster and wish to consume from the node that hosts the queue.** -**See https://docs.spring.io/spring-amqp/reference/html/_reference.html#queue-affinity[Queue Affinity and the LocalizedQueueConnectionFactory] for more information.** +Only needed if you use a RabbitMQ cluster and wish to consume from the node that hosts the queue. +See https://docs.spring.io/spring-amqp/reference/html/_reference.html#queue-affinity[Queue Affinity and the LocalizedQueueConnectionFactory] for more information. + Default: empty. spring.cloud.stream.rabbit.binder.nodes:: - A comma-separated list of RabbitMQ node names. +A comma-separated list of RabbitMQ node names. When more than one entry, used to locate the server address where a queue is located. Each entry in this list must have a corresponding entry in `spring.rabbitmq.addresses`. -**Only needed if you are using a RabbitMQ cluster and wish to consume from the node that hosts the queue.** -**See https://docs.spring.io/spring-amqp/reference/html/_reference.html#queue-affinity[Queue Affinity and the LocalizedQueueConnectionFactory] for more information.** +Only needed if you use a RabbitMQ cluster and wish to consume from the node that hosts the queue. +See https://docs.spring.io/spring-amqp/reference/html/_reference.html#queue-affinity[Queue Affinity and the LocalizedQueueConnectionFactory] for more information. + Default: empty. spring.cloud.stream.rabbit.binder.compressionLevel:: - Compression level for compressed bindings. +The compression level for compressed bindings. See `java.util.zip.Deflater`. + Default: `1` (BEST_LEVEL). spring.cloud.stream.binder.connection-name-prefix:: - A connection name prefix used to name the connection(s) created by this binder. -The name will be this prefix followed by `#n`, where n increments each time a new connection is opened. +A connection name prefix used to name the connection(s) created by this binder. +The name is this prefix followed by `#n`, where `n` increments each time a new connection is opened. + Default: none (Spring AMQP default). === RabbitMQ Consumer Properties -The following properties are available for Rabbit consumers only and -must be prefixed with `spring.cloud.stream.rabbit.bindings..consumer.`. +The following properties are available for Rabbit consumers only and must be prefixed with `spring.cloud.stream.rabbit.bindings..consumer.`. acknowledgeMode:: - The acknowledge mode. +The acknowledge mode. + Default: `AUTO`. autoBindDlq:: - Whether to automatically declare the DLQ and bind it to the binder DLX. +Whether to automatically declare the DLQ and bind it to the binder DLX. + Default: `false`. bindingRoutingKey:: - The routing key with which to bind the queue to the exchange (if `bindQueue` is `true`). - for partitioned destinations `-` will be appended. +The routing key with which to bind the queue to the exchange (if `bindQueue` is `true`). +For partitioned destinations, `-` is appended. + Default: `#`. bindQueue:: - Whether to bind the queue to the destination exchange; set to `false` if you have set up your own infrastructure and have previously created/bound the queue. +Whether to bind the queue to the destination exchange. +Set it to `false` if you have set up your own infrastructure and have previously created and bound the queue. + Default: `true`. deadLetterQueueName:: - name of the DLQ +The name of the DLQ + Default: `prefix+destination.dlq` deadLetterExchange:: - a DLX to assign to the queue; if autoBindDlq is true +A DLX to assign to the queue. +Relevant only if `autoBindDlq` is `true`. + Default: 'prefix+DLX' deadLetterRoutingKey:: - a dead letter routing key to assign to the queue; if autoBindDlq is true +A dead letter routing key to assign to the queue. +Relevant only if `autoBindDlq` is `true`. + Default: `destination` declareExchange:: - Whether to declare the exchange for the destination. +Whether to declare the exchange for the destination. + Default: `true`. delayedExchange:: - Whether to declare the exchange as a `Delayed Message Exchange` - requires the delayed message exchange plugin on the broker. - The `x-delayed-type` argument is set to the `exchangeType`. +Whether to declare the exchange as a `Delayed Message Exchange`. +Requires the delayed message exchange plugin on the broker. +The `x-delayed-type` argument is set to the `exchangeType`. + Default: `false`. dlqDeadLetterExchange:: - if a DLQ is declared, a DLX to assign to that queue +If a DLQ is declared, a DLX to assign to that queue. + Default: `none` dlqDeadLetterRoutingKey:: - if a DLQ is declared, a dead letter routing key to assign to that queue; default none +If a DLQ is declared, a dead letter routing key to assign to that queue. + Default: `none` dlqExpires:: - how long before an unused dead letter queue is deleted (ms) +How long before an unused dead letter queue is deleted (in milliseconds). + Default: `no expiration` dlqLazy:: - Declare the dead letter queue with the `x-queue-mode=lazy` argument. - See https://www.rabbitmq.com/lazy-queues.html[Lazy Queues]. - Consider using a policy instead of this setting because using a policy allows changing the setting without deleting the queue. +Declare the dead letter queue with the `x-queue-mode=lazy` argument. +See https://www.rabbitmq.com/lazy-queues.html["`Lazy Queues`"]. +Consider using a policy instead of this setting, because using a policy allows changing the setting without deleting the queue. + Default: `false`. dlqMaxLength:: - maximum number of messages in the dead letter queue +Maximum number of messages in the dead letter queue. + Default: `no limit` dlqMaxLengthBytes:: - maximum number of total bytes in the dead letter queue from all messages +Maximum number of total bytes in the dead letter queue from all messages. + Default: `no limit` dlqMaxPriority:: - maximum priority of messages in the dead letter queue (0-255) +Maximum priority of messages in the dead letter queue (0-255). + Default: `none` dlqTtl:: - default time to live to apply to the dead letter queue when declared (ms) +Default time to live to apply to the dead letter queue when declared (in milliseconds). + Default: `no limit` durableSubscription:: - Whether subscription should be durable. +Whether the subscription should be durable. Only effective if `group` is also set. + Default: `true`. exchangeAutoDelete:: - If `declareExchange` is true, whether the exchange should be auto-delete (removed after the last queue is removed). +If `declareExchange` is true, whether the exchange should be auto-deleted (that is, removed after the last queue is removed). + Default: `true`. exchangeDurable:: - If `declareExchange` is true, whether the exchange should be durable (survives broker restart). +If `declareExchange` is true, whether the exchange should be durable (that is, it survives broker restart). + Default: `true`. exchangeType:: - The exchange type; `direct`, `fanout` or `topic` for non-partitioned destinations; `direct` or `topic` for partitioned destinations. +The exchange type: `direct`, `fanout` or `topic` for non-partitioned destinations and `direct` or `topic` for partitioned destinations. + Default: `topic`. exclusive:: - Create an exclusive consumer; concurrency should be 1 when this is `true`; often used when strict ordering is required but enabling a hot standby instance to take over after a failure. - See `recoveryInterval`, which controls how often a standby instance will attempt to consume. +Whether to create an exclusive consumer. +Concurrency should be 1 when this is `true`. +Often used when strict ordering is required but enabling a hot standby instance to take over after a failure. +See `recoveryInterval`, which controls how often a standby instance attempts to consume. + Default: `false`. expires:: - how long before an unused queue is deleted (ms) +How long before an unused queue is deleted (in milliseconds). + Default: `no expiration` failedDeclarationRetryInterval:: - The interval (ms) between attempts to consume from a queue if it is missing. +The interval (in milliseconds) between attempts to consume from a queue if it is missing. + Default: 5000 headerPatterns:: - Patterns for headers to be mapped from inbound messages. +Patterns for headers to be mapped from inbound messages. + Default: `['*']` (all headers). lazy:: - Declare the queue with the `x-queue-mode=lazy` argument. - See https://www.rabbitmq.com/lazy-queues.html[Lazy Queues]. - Consider using a policy instead of this setting because using a policy allows changing the setting without deleting the queue. +Declare the queue with the `x-queue-mode=lazy` argument. +See https://www.rabbitmq.com/lazy-queues.html["`Lazy Queues`"]. +Consider using a policy instead of this setting, because using a policy allows changing the setting without deleting the queue. + Default: `false`. maxConcurrency:: - the maximum number of consumers +The maximum number of consumers. + Default: `1`. maxLength:: - maximum number of messages in the queue +The maximum number of messages in the queue. + Default: `no limit` maxLengthBytes:: - maximum number of total bytes in the queue from all messages +The maximum number of total bytes in the queue from all messages. + Default: `no limit` maxPriority:: - maximum priority of messages in the queue (0-255) +The maximum priority of messages in the queue (0-255). + Default: `none` missingQueuesFatal:: - If the queue cannot be found, treat the condition as fatal and stop the listener container. - Defaults to `false` so that the container keeps trying to consume from the queue, for example when using a cluster and the node hosting a non HA queue is down. +When the queue cannot be found, whether to treat the condition as fatal and stop the listener container. +Defaults to `false` so that the container keeps trying to consume from the queue -- for example, when using a cluster and the node hosting a non-HA queue is down. + Default: `false` prefetch:: - Prefetch count. +Prefetch count. + Default: `1`. prefix:: - A prefix to be added to the name of the `destination` and queues. +A prefix to be added to the name of the `destination` and queues. + Default: "". queueDeclarationRetries:: - The number of times to retry consuming from a queue if it is missing. - Only relevant if `missingQueuesFatal` is `true`; otherwise the container keeps retrying indefinitely. +The number of times to retry consuming from a queue if it is missing. +Relevant only when `missingQueuesFatal` is `true`. +Otherwise, the container keeps retrying indefinitely. + Default: `3` queueNameGroupOnly:: - When true, consume from a queue with a name equal to the `group`; otherwise the queue name is `destination.group`. - This is useful, for example, when using Spring Cloud Stream to consume from an existing RabbitMQ queue. +When true, consume from a queue with a name equal to the `group`. +Otherwise the queue name is `destination.group`. +This is useful, for example, when using Spring Cloud Stream to consume from an existing RabbitMQ queue. + Default: false. recoveryInterval:: - The interval between connection recovery attempts, in milliseconds. +The interval between connection recovery attempts, in milliseconds. + Default: `5000`. requeueRejected:: - Whether delivery failures should be requeued when retry is disabled or republishToDlq is false. +Whether delivery failures should be re-queued when retry is disabled or `republishToDlq` is `false`. + Default: `false`. +[[spring-cloud-stream-rabbit-republish-delivery-mode]] republishDeliveryMode:: - When `republishToDlq` is `true`, specify the delivery mode of the republished message. +When `republishToDlq` is `true`, specifies the delivery mode of the republished message. + Default: `DeliveryMode.PERSISTENT` republishToDlq:: - By default, messages which fail after retries are exhausted are rejected. -If a dead-letter queue (DLQ) is configured, RabbitMQ will route the failed message (unchanged) to the DLQ. -If set to `true`, the binder will republish failed messages to the DLQ with additional headers, including the exception message and stack trace from the cause of the final failure. +By default, messages that fail after retries are exhausted are rejected. +If a dead-letter queue (DLQ) is configured, RabbitMQ routes the failed message (unchanged) to the DLQ. +If set to `true`, the binder republishs failed messages to the DLQ with additional headers, including the exception message and stack trace from the cause of the final failure. + Default: false transacted:: - Whether to use transacted channels. +Whether to use transacted channels. + Default: `false`. ttl:: -default time to live to apply to the queue when declared (ms) +Default time to live to apply to the queue when declared (in milliseconds). + Default: `no limit` txSize:: - The number of deliveries between acks. +The number of deliveries between acks. + Default: `1`. @@ -299,212 +307,219 @@ The following properties are available for Rabbit producers only and must be prefixed with `spring.cloud.stream.rabbit.bindings..producer.`. autoBindDlq:: - Whether to automatically declare the DLQ and bind it to the binder DLX. +Whether to automatically declare the DLQ and bind it to the binder DLX. + Default: `false`. batchingEnabled:: - Whether to enable message batching by producers. -Messages are batched into one message according to the following properties. -Refer to https://docs.spring.io/spring-amqp//reference/html/_reference.html#template-batching[Batching] for more information. +Whether to enable message batching by producers. +Messages are batched into one message according to the following properties (described in the next three entries in this list): 'batchSize', `batchBufferLimit`, and `batchTimeout`. +See https://docs.spring.io/spring-amqp//reference/html/_reference.html#template-batching[Batching] for more information. + Default: `false`. batchSize:: - The number of messages to buffer when batching is enabled. +The number of messages to buffer when batching is enabled. + Default: `100`. batchBufferLimit:: - The maximum buffer size when batching is enabled. +The maximum buffer size when batching is enabled. + - Default: `10000`. +Default: `10000`. batchTimeout:: - The batch timeout when batching is enabled. +The batch timeout when batching is enabled. + - Default: `5000`. +Default: `5000`. bindingRoutingKey:: - The routing key with which to bind the queue to the exchange (if `bindQueue` is `true`). - Only applies to non-partitioned destinations. - Only applies if `requiredGroups` are provided and then only to those groups. +The routing key with which to bind the queue to the exchange (if `bindQueue` is `true`). +Only applies to non-partitioned destinations. +Only applies if `requiredGroups` are provided and then only to those groups. + Default: `#`. bindQueue:: - Whether to bind the queue to the destination exchange; set to `false` if you have set up your own infrastructure and have previously created/bound the queue. - Only applies if `requiredGroups` are provided and then only to those groups. +Whether to bind the queue to the destination exchange. +Set it to `false` if you have set up your own infrastructure and have previously created and bound the queue. +Only applies if `requiredGroups` are provided and then only to those groups. + Default: `true`. compress:: - Whether data should be compressed when sent. +Whether data should be compressed when sent. + Default: `false`. deadLetterQueueName:: - name of the DLQ - Only applies if `requiredGroups` are provided and then only to those groups. +The name of the DLQ +Only applies if `requiredGroups` are provided and then only to those groups. + Default: `prefix+destination.dlq` deadLetterExchange:: - a DLX to assign to the queue; if autoBindDlq is true - Only applies if `requiredGroups` are provided and then only to those groups. +A DLX to assign to the queue. +Relevant only when `autoBindDlq` is `true`. +Applies only when `requiredGroups` are provided and then only to those groups. + Default: 'prefix+DLX' deadLetterRoutingKey:: - a dead letter routing key to assign to the queue; if autoBindDlq is true - Only applies if `requiredGroups` are provided and then only to those groups. +A dead letter routing key to assign to the queue. +Relevant only when `autoBindDlq` is `true`. +Applies only when `requiredGroups` are provided and then only to those groups. + Default: `destination` declareExchange:: - Whether to declare the exchange for the destination. +Whether to declare the exchange for the destination. + Default: `true`. delayExpression:: - A SpEL expression to evaluate the delay to apply to the message (`x-delay` header) - has no effect if the exchange is not a delayed message exchange. +A SpEL expression to evaluate the delay to apply to the message (`x-delay` header). +It has no effect if the exchange is not a delayed message exchange. + Default: No `x-delay` header is set. delayedExchange:: - Whether to declare the exchange as a `Delayed Message Exchange` - requires the delayed message exchange plugin on the broker. - The `x-delayed-type` argument is set to the `exchangeType`. +Whether to declare the exchange as a `Delayed Message Exchange`. +Requires the delayed message exchange plugin on the broker. +The `x-delayed-type` argument is set to the `exchangeType`. + Default: `false`. deliveryMode:: - Delivery mode. +The delivery mode. + Default: `PERSISTENT`. dlqDeadLetterExchange:: - if a DLQ is declared, a DLX to assign to that queue - Only applies if `requiredGroups` are provided and then only to those groups. +When a DLQ is declared, a DLX to assign to that queue. +Applies only if `requiredGroups` are provided and then only to those groups. + Default: `none` dlqDeadLetterRoutingKey:: - if a DLQ is declared, a dead letter routing key to assign to that queue; default none - Only applies if `requiredGroups` are provided and then only to those groups. +When a DLQ is declared, a dead letter routing key to assign to that queue. +Applies only when `requiredGroups` are provided and then only to those groups. + Default: `none` dlqExpires:: - how long before an unused dead letter queue is deleted (ms) - Only applies if `requiredGroups` are provided and then only to those groups. +How long (in milliseconds) before an unused dead letter queue is deleted. +Applies only when `requiredGroups` are provided and then only to those groups. + Default: `no expiration` dlqLazy:: - Declare the dead letter queue with the `x-queue-mode=lazy` argument. - See https://www.rabbitmq.com/lazy-queues.html[Lazy Queues]. - Consider using a policy instead of this setting because using a policy allows changing the setting without deleting the queue. - Only applies if `requiredGroups` are provided and then only to those groups. +Declare the dead letter queue with the `x-queue-mode=lazy` argument. +See https://www.rabbitmq.com/lazy-queues.html["`Lazy Queues`"]. +Consider using a policy instead of this setting, because using a policy allows changing the setting without deleting the queue. +Applies only when `requiredGroups` are provided and then only to those groups. + dlqMaxLength:: - maximum number of messages in the dead letter queue - Only applies if `requiredGroups` are provided and then only to those groups. +Maximum number of messages in the dead letter queue. +Applies only if `requiredGroups` are provided and then only to those groups. + Default: `no limit` dlqMaxLengthBytes:: - maximum number of total bytes in the dead letter queue from all messages - Only applies if `requiredGroups` are provided and then only to those groups. +Maximum number of total bytes in the dead letter queue from all messages. +Applies only when `requiredGroups` are provided and then only to those groups. + Default: `no limit` dlqMaxPriority:: - maximum priority of messages in the dead letter queue (0-255) - Only applies if `requiredGroups` are provided and then only to those groups. +Maximum priority of messages in the dead letter queue (0-255) +Applies only when `requiredGroups` are provided and then only to those groups. + Default: `none` dlqTtl:: - default time to live to apply to the dead letter queue when declared (ms) - Only applies if `requiredGroups` are provided and then only to those groups. +Default time (in milliseconds) to live to apply to the dead letter queue when declared. +Applies only when `requiredGroups` are provided and then only to those groups. + Default: `no limit` exchangeAutoDelete:: - If `declareExchange` is true, whether the exchange should be auto-delete (removed after the last queue is removed). +If `declareExchange` is `true`, whether the exchange should be auto-delete (it is removed after the last queue is removed). + Default: `true`. exchangeDurable:: - If `declareExchange` is true, whether the exchange should be durable (survives broker restart). +If `declareExchange` is `true`, whether the exchange should be durable (survives broker restart). + Default: `true`. exchangeType:: - The exchange type; `direct`, `fanout` or `topic` for non-partitioned destinations; `direct` or `topic` for partitioned destinations. +The exchange type: `direct`, `fanout` or `topic` for non-partitioned destinations and `direct` or `topic` for partitioned destinations. + Default: `topic`. expires:: - how long before an unused queue is deleted (ms) - Only applies if `requiredGroups` are provided and then only to those groups. +How long (in milliseconds) before an unused queue is deleted. +Applies only when `requiredGroups` are provided and then only to those groups. + Default: `no expiration` headerPatterns:: - Patterns for headers to be mapped to outbound messages. +Patterns for headers to be mapped to outbound messages. + Default: `['*']` (all headers). lazy:: - Declare the queue with the `x-queue-mode=lazy` argument. - See https://www.rabbitmq.com/lazy-queues.html[Lazy Queues]. - Consider using a policy instead of this setting because using a policy allows changing the setting without deleting the queue. - Only applies if `requiredGroups` are provided and then only to those groups. +Declare the queue with the `x-queue-mode=lazy` argument. +See https://www.rabbitmq.com/lazy-queues.html["`Lazy Queues`"]. +Consider using a policy instead of this setting, because using a policy allows changing the setting without deleting the queue. +Applies only when `requiredGroups` are provided and then only to those groups. + Default: `false`. maxLength:: - maximum number of messages in the queue - Only applies if `requiredGroups` are provided and then only to those groups. +Maximum number of messages in the queue. +Applies only when `requiredGroups` are provided and then only to those groups. + Default: `no limit` maxLengthBytes:: - maximum number of total bytes in the queue from all messages - Only applies if `requiredGroups` are provided and then only to those groups. +Maximum number of total bytes in the queue from all messages. +Only applies if `requiredGroups` are provided and then only to those groups. + Default: `no limit` maxPriority:: - maximum priority of messages in the queue (0-255) - Only applies if `requiredGroups` are provided and then only to those groups. +Maximum priority of messages in the queue (0-255). +Only applies if `requiredGroups` are provided and then only to those groups. + Default: `none` prefix:: - A prefix to be added to the name of the `destination` exchange. +A prefix to be added to the name of the `destination` exchange. + Default: "". queueNameGroupOnly:: - When true, consume from a queue with a name equal to the `group`; otherwise the queue name is `destination.group`. - This is useful, for example, when using Spring Cloud Stream to consume from an existing RabbitMQ queue. - Only applies if `requiredGroups` are provided and then only to those groups. +When `true`, consume from a queue with a name equal to the `group`. +Otherwise the queue name is `destination.group`. +This is useful, for example, when using Spring Cloud Stream to consume from an existing RabbitMQ queue. +Applies only when `requiredGroups` are provided and then only to those groups. + Default: false. routingKeyExpression:: - A SpEL expression to determine the routing key to use when publishing messages. - For a fixed routing key, use a literal expression, e.g. `routingKeyExpression='my.routingKey'` in a properties file, or `routingKeyExpression: '''my.routingKey'''` in a YAML file. +A SpEL expression to determine the routing key to use when publishing messages. +For a fixed routing key, use a literal expression, such as `routingKeyExpression='my.routingKey'` in a properties file or `routingKeyExpression: '''my.routingKey'''` in a YAML file. + Default: `destination` or `destination-` for partitioned destinations. transacted:: - Whether to use transacted channels. +Whether to use transacted channels. + Default: `false`. ttl:: - default time to live to apply to the queue when declared (ms) - Only applies if `requiredGroups` are provided and then only to those groups. +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` -[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). -==== +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. == Retry With the RabbitMQ Binder -=== Overview - When retry is enabled within the binder, the listener container thread is suspended for any back off periods that are configured. -This might be important when strict ordering is required with a single consumer but for other use cases it prevents other messages from being processed on that thread. -An alternative to using binder retry is to set up dead lettering with time to live on the dead-letter queue (DLQ), as well as dead-letter configuration on the DLQ itself. -See <> for more information about the properties discussed here. -Example configuration to enable this feature: +This might be important when strict ordering is required with a single consumer. However, for other use cases, it prevents other messages from being processed on that thread. +An alternative to using binder retry is to set up dead lettering with time to live on the dead-letter queue (DLQ) as well as dead-letter configuration on the DLQ itself. +See "`<>`" for more information about the properties discussed here. +You can use the following example configuration to enable this feature: -* Set `autoBindDlq` to `true` - the binder will create a DLQ; you can optionally specify a name in `deadLetterQueueName` -* Set `dlqTtl` to the back off time you want to wait between redeliveries -* Set the `dlqDeadLetterExchange` to the default exchange - expired messages from the DLQ will be routed to the original queue since the default `deadLetterRoutingKey` is the queue name (`destination.group`) - setting to the default exchange is achieved by setting the property with no value, as is shown in the example below +* Set `autoBindDlq` to `true`. +The binder create a DLQ. +Optionally, you can specify a name in `deadLetterQueueName`. +* Set `dlqTtl` to the back off time you want to wait between redeliveries. +* Set the `dlqDeadLetterExchange` to the default exchange. +Expired messages from the DLQ are routed to the original queue, because the default `deadLetterRoutingKey` is the queue name (`destination.group`). +Setting to the default exchange is achieved by setting the property with no value, as shown in the next example. -To force a message to be dead-lettered, either throw an `AmqpRejectAndDontRequeueException`, or set `requeueRejected` to `true` (default) and throw any exception. +To force a message to be dead-lettered, either throw an `AmqpRejectAndDontRequeueException` or set `requeueRejected` to `true` (the default) and throw any exception. -The loop will continue without end, which is fine for transient problems but you may want to give up after some number of attempts. -Fortunately, RabbitMQ provides the `x-death` header which allows you to determine how many cycles have occurred. +The loop continue without end, which is fine for transient problems, but you may want to give up after some number of attempts. +Fortunately, RabbitMQ provides the `x-death` header, which lets you determine how many cycles have occurred. To acknowledge a message after giving up, throw an `ImmediateAcknowledgeAmqpException`. === Putting it All Together +The following configuration creates an exchange `myDestination` with queue `myDestination.consumerGroup` bound to a topic exchange with a wildcard routing key `#`: + [source] --- spring.cloud.stream.bindings.input.destination=myDestination @@ -517,10 +532,9 @@ spring.cloud.stream.rabbit.bindings.input.consumer.dlq-ttl=5000 spring.cloud.stream.rabbit.bindings.input.consumer.dlq-dead-letter-exchange= --- -This configuration creates an exchange `myDestination` with queue `myDestination.consumerGroup` bound to a topic exchange with a wildcard routing key `#`. -It creates a DLQ bound to a direct exchange `DLX` with routing key `myDestination.consumerGroup`. +This configuration creates a DLQ bound to a direct exchange (`DLX`) with a routing key of `myDestination.consumerGroup`. When messages are rejected, they are routed to the DLQ. -After 5 seconds, the message expires and is routed to the original queue using the queue name as the routing key. +After 5 seconds, the message expires and is routed to the original queue by using the queue name as the routing key, as shown in the following example: .Spring Boot application [source, java] @@ -550,38 +564,40 @@ Notice that the count property in the `x-death` header is a `Long`. [[rabbit-error-channels]] == Error Channels -Starting with _version 1.3_, the binder unconditionally sends exceptions to an error channel for each consumer destination, and can be configured to send async producer send failures to an error channel too. -See <> for more information. +Starting with version 1.3, the binder unconditionally sends exceptions to an error channel for each consumer destination and can also be configured to send async producer send failures to an error channel. +See "`<>`" for more information. -With rabbitmq, there are two types of send failures: +RabbitMQ has two types of send failures: -* returned messages -* negatively acknowledged https://www.rabbitmq.com/confirms.html[Publisher Confirms] +* Returned messages, +* Negatively acknowledged https://www.rabbitmq.com/confirms.html[Publisher Confirms]. -The latter is rare; quoting the RabbitMQ documentation "[A nack] will only be delivered if an internal error occurs in the Erlang process responsible for a queue.". +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.". -As well as enabling producer error channels as described in <>, the RabbitMQ binder will only send messages to the channels if the connection factory is appropriately configured: +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: * `ccf.setPublisherConfirms(true);` * `ccf.setPublisherReturns(true);` -When using spring boot configuration for the connection factory, set properties: +When using Spring Boot configuration for the connection factory, set the following properties: * `spring.rabbitmq.publisher-confirms` * `spring.rabbitmq.publisher-returns` -The payload of the `ErrorMessage` for a returned message is a `ReturnedAmqpMessageException` with properties: +The payload of the `ErrorMessage` for a returned message is a `ReturnedAmqpMessageException` with the following properties: -* `failedMessage` - the spring-messaging `Message` that failed to be sent. -* `amqpMessage` - the raw spring-amqp `Message` -* `replyCode` - an integer value indicating the reason for the failure (e.g. 312 - No route) -* `replyText` - a text value indicating the reason for the failure e.g. `NO_ROUTE`. -* `exchange` - the exchange to which the message was published. -* `routingKey` - the routing key used when the message was published. +* `failedMessage`: The spring-messaging `Message` that failed to be sent. +* `amqpMessage`: The raw spring-amqp `Message`. +* `replyCode`: An integer value indicating the reason for the failure (for example, 312 - No route). +* `replyText`: A text value indicating the reason for the failure (for example, `NO_ROUTE`). +* `exchange`: The exchange to which the message was published. +* `routingKey`: The routing key used when the message was published. -For negatively acknowledged confirms, the payload is a `NackedAmqpMessageException` with properties: +For negatively acknowledged confirmations, the payload is a `NackedAmqpMessageException` with the following properties: -* `failedMessage` - the spring-messaging `Message` that failed to be sent. -* `nackReason` - a reason (if available; you may need to examine the broker logs for more information). +* `failedMessage`: The spring-messaging `Message` that failed to be sent. +* `nackReason`: A reason (if available -- you may need to examine the broker logs for more information). -There is no automatic handling of these exceptions (such as sending to a <>); you can consume these exceptions with your own Spring Integration flow. +There is no automatic handling of these exceptions (such as sending to a <>). +You can consume these exceptions with your own Spring Integration flow. diff --git a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/partitions.adoc b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/partitions.adoc index 8099dff46..0be97700f 100644 --- a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/partitions.adoc +++ b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/partitions.adoc @@ -2,11 +2,11 @@ RabbitMQ does not support partitioning natively. -Sometimes it is advantageous to send data to specific partitions, for example when you want to strictly order message processing - all messages for a particular customer should go to the same partition. +Sometimes, it is advantageous to send data to specific partitions -- for example, when you want to strictly order message processing, all messages for a particular customer should go to the same partition. The `RabbitMessageChannelBinder` provides partitioning by binding a queue for each partition to the destination exchange. -The following illustrates how to configure the producer and consumer side: +The following Java and YAML examples show how to configure the producer: .Producer [source, java] @@ -18,10 +18,10 @@ public class RabbitPartitionProducerApplication { private static final Random RANDOM = new Random(System.currentTimeMillis()); private static final String[] data = new String[] { - "foo1", "bar1", "qux1", - "foo2", "bar2", "qux2", - "foo3", "bar3", "qux3", - "foo4", "bar4", "qux4", + "abc1", "def1", "qux1", + "abc2", "def2", "qux2", + "abc3", "def3", "qux3", + "abc4", "def4", "qux4", }; public static void main(String[] args) { @@ -61,25 +61,28 @@ public class RabbitPartitionProducerApplication { [NOTE] ==== -The above configuration uses the default partitioning (`key.hashCode() % partitionCount`). -This may or may not provide a suitably balanced algorithm, depending on the key values; you can override this default by using the `partitionSelectorExpression` or `partitionSelectorClass` properties. +The configuration in the prececing example uses the default partitioning (`key.hashCode() % partitionCount`). +This may or may not provide a suitably balanced algorithm, depending on the key values. +You can override this default by using the `partitionSelectorExpression` or `partitionSelectorClass` properties. -The `required-groups` property is only required if you need the consumer queues to be provisioned when the producer is deployed. -Otherwise, any messages sent to a partition will be lost until the corresponding consumer is deployed. +The `required-groups` property is required only if you need the consumer queues to be provisioned when the producer is deployed. +Otherwise, any messages sent to a partition are lost until the corresponding consumer is deployed. ==== -This configuration provisions a topic exchange: +The following configuration provisions a topic exchange: image::part-exchange.png[scaledwidth="50%"] -and these queues bound to that exchange: +The following queues are bound to that exchange: image::part-queues.png[scaledwidth="50%"] -with these bindings: +The following bindings associate the queues to the exchange: image::part-bindings.png[scaledwidth="50%"] +The following Java and YAML examples continue the previous examples and show how to configure the consumer: + .Consumer [source, java] ---- @@ -116,6 +119,7 @@ public class RabbitPartitionConsumerApplication { instance-index: 0 ---- -IMPORTANT: The `RabbitMessageChannelBinder` does not support dynamic scaling; there must be at least one consumer per partition. -The consumer's `instanceIndex` is used to indicate which partition will be consumed. -On platforms such as Cloud Foundry there can only be one instance with an `instanceIndex`. +IMPORTANT: The `RabbitMessageChannelBinder` does not support dynamic scaling. +There must be at least one consumer per partition. +The consumer's `instanceIndex` is used to indicate which partition is consumed. +Platforms such as Cloud Foundry can have only one instance with an `instanceIndex`.