SpringCloud/spring-cloud-static
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
Files
05e3a6afeecb344de577fd00a4eb990e7e49e593
spring-cloud-static/Edgware.RELEASE/multi/multi__rabbitmq_binder.html
Spencer Gibb fc5375a2e0 Sync docs from vEdgware.RELEASE to gh-pages
2017-11-22 14:01:35 -05:00

304 lines
65 KiB
HTML
Raw Blame History

<html><head>
<meta http-equiv="Content-Type" content="text/html; charset=ISO-8859-1">
<title>38.&nbsp;RabbitMQ Binder</title><link rel="stylesheet" type="text/css" href="css/manual-multipage.css"><meta name="generator" content="DocBook XSL Stylesheets V1.78.1"><link rel="home" href="multi_spring-cloud.html" title="Spring Cloud"><link rel="up" href="multi__binder_implementations.html" title="Part&nbsp;V.&nbsp;Binder Implementations"><link rel="prev" href="multi__apache_kafka_binder.html" title="37.&nbsp;Apache Kafka Binder"><link rel="next" href="multi__spring_cloud_bus.html" title="Part&nbsp;VI.&nbsp;Spring Cloud Bus"></head><body bgcolor="white" text="black" link="#0000FF" vlink="#840084" alink="#0000FF"><div class="navheader"><table width="100%" summary="Navigation header"><tr><th colspan="3" align="center">38.&nbsp;RabbitMQ Binder</th></tr><tr><td width="20%" align="left"><a accesskey="p" href="multi__apache_kafka_binder.html">Prev</a>&nbsp;</td><th width="60%" align="center">Part&nbsp;V.&nbsp;Binder Implementations</th><td width="20%" align="right">&nbsp;<a accesskey="n" href="multi__spring_cloud_bus.html">Next</a></td></tr></table><hr></div><div class="chapter"><div class="titlepage"><div><div><h2 class="title"><a name="_rabbitmq_binder" href="#_rabbitmq_binder"></a>38.&nbsp;RabbitMQ Binder</h2></div></div></div><div class="section"><div class="titlepage"><div><div><h2 class="title" style="clear: both"><a name="_usage_2" href="#_usage_2"></a>38.1&nbsp;Usage</h2></div></div></div><p>For using the RabbitMQ binder, you just need to add it to your Spring Cloud Stream application, using the following Maven coordinates:</p><pre class="programlisting"><span xmlns:d="http://docbook.org/ns/docbook" class="hl-tag">&lt;dependency&gt;</span>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-tag">&lt;groupId&gt;</span>org.springframework.cloud<span xmlns:d="http://docbook.org/ns/docbook" class="hl-tag">&lt;/groupId&gt;</span>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-tag">&lt;artifactId&gt;</span>spring-cloud-stream-binder-rabbit<span xmlns:d="http://docbook.org/ns/docbook" class="hl-tag">&lt;/artifactId&gt;</span>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-tag">&lt;/dependency&gt;</span></pre><p>Alternatively, you can also use the Spring Cloud Stream RabbitMQ Starter.</p><pre class="programlisting"><span xmlns:d="http://docbook.org/ns/docbook" class="hl-tag">&lt;dependency&gt;</span>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-tag">&lt;groupId&gt;</span>org.springframework.cloud<span xmlns:d="http://docbook.org/ns/docbook" class="hl-tag">&lt;/groupId&gt;</span>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-tag">&lt;artifactId&gt;</span>spring-cloud-starter-stream-rabbit<span xmlns:d="http://docbook.org/ns/docbook" class="hl-tag">&lt;/artifactId&gt;</span>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-tag">&lt;/dependency&gt;</span></pre></div><div class="section"><div class="titlepage"><div><div><h2 class="title" style="clear: both"><a name="_rabbitmq_binder_overview" href="#_rabbitmq_binder_overview"></a>38.2&nbsp;RabbitMQ Binder Overview</h2></div></div></div><p>A simplified diagram of how the RabbitMQ binder operates can be seen below.</p><div class="figure"><a name="d0e10332" href="#d0e10332"></a><p class="title"><b>Figure&nbsp;38.1.&nbsp;RabbitMQ Binder</b></p><div class="figure-contents"><div class="mediaobject"><img src="images/rabbit-binder.png" alt="rabbit binder"></div></div></div><br class="figure-break"><p>The RabbitMQ Binder implementation maps each destination to a <code class="literal">TopicExchange</code>.
For each consumer group, a <code class="literal">Queue</code> will be bound to that <code class="literal">TopicExchange</code>.
Each consumer instance have a corresponding RabbitMQ <code class="literal">Consumer</code> instance for its group&#8217;s <code class="literal">Queue</code>.
For partitioned producers/consumers the queues are suffixed with the partition index and use the partition index as routing key.</p><p>Using the <code class="literal">autoBindDlq</code> option, you can optionally configure the binder to create and configure dead-letter queues (DLQs) (and a dead-letter exchange <code class="literal">DLX</code>).
The dead letter queue has the name of the destination, appended with <code class="literal">.dlq</code>.
If retry is enabled (<code class="literal">maxAttempts &gt; 1</code>) failed messages will be delivered to the DLQ.
If retry is disabled (<code class="literal">maxAttempts = 1</code>), you should set <code class="literal">requeueRejected</code> to <code class="literal">false</code> (default) so that a failed message will be routed to the DLQ, instead of being requeued.
In addition, <code class="literal">republishToDlq</code> 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 <code class="literal">x-exception-stacktrace</code> header.
This option does not need retry enabled; you can republish a failed message after just one attempt.
Starting with <span class="emphasis"><em>version 1.2</em></span>, you can configure the delivery mode of republished messages; see property <code class="literal">republishDeliveryMode</code>.</p><div class="important" style="margin-left: 0.5in; margin-right: 0.5in;"><table border="0" summary="Important"><tr><td rowspan="2" align="center" valign="top" width="25"><img alt="[Important]" src="images/important.png"></td><th align="left">Important</th></tr><tr><td align="left" valign="top"><p>Setting <code class="literal">requeueRejected</code> to <code class="literal">true</code> will cause the message to be requeued and redelivered continually, which is likely not what you want unless the failure issue is transient.
In general, it&#8217;s better to enable retry within the binder by setting <code class="literal">maxAttempts</code> to greater than one, or set <code class="literal">republishToDlq</code> to <code class="literal">true</code>.</p></td></tr></table></div><p>See <a class="xref" href="multi__rabbitmq_binder.html#rabbit-binder-properties" title="38.3.1&nbsp;RabbitMQ Binder Properties">Section&nbsp;38.3.1, &#8220;RabbitMQ Binder Properties&#8221;</a> for more information about these properties.</p><p>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 <a class="xref" href="multi__rabbitmq_binder.html#rabbit-dlq-processing" title="38.6&nbsp;Dead-Letter Queue Processing">Section&nbsp;38.6, &#8220;Dead-Letter Queue Processing&#8221;</a>.</p><div class="note" style="margin-left: 0.5in; margin-right: 0.5in;"><table border="0" summary="Note"><tr><td rowspan="2" align="center" valign="top" width="25"><img alt="[Note]" src="images/note.png"></td><th align="left">Note</th></tr><tr><td align="left" valign="top"><p>When <span class="strong"><strong>multiple</strong></span> RabbitMQ binders are used in a Spring Cloud Stream application, it is important to disable 'RabbitAutoConfiguration' to avoid the same configuration from <code class="literal">RabbitAutoConfiguration</code> being applied to the two binders.</p></td></tr></table></div><p>Starting with <span class="emphasis"><em>version 1.3</em></span>, the <code class="literal">RabbitMessageChannelBinder</code> creates an internal <code class="literal">ConnectionFactory</code> copy for the non-transactional producers to avoid dead locks on consumers when shared, cached connections are blocked because of <a class="link" href="https://www.rabbitmq.com/memory.html" target="_top">Memory Alarm</a> on Broker.</p></div><div class="section"><div class="titlepage"><div><div><h2 class="title" style="clear: both"><a name="_configuration_options_3" href="#_configuration_options_3"></a>38.3&nbsp;Configuration Options</h2></div></div></div><p>This section contains settings specific to the RabbitMQ Binder and bound channels.</p><p>For general binding configuration options and properties,
please refer to the <a class="link" href="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" target="_top">Spring Cloud Stream core documentation</a>.</p><div class="section"><div class="titlepage"><div><div><h3 class="title"><a name="rabbit-binder-properties" href="#rabbit-binder-properties"></a>38.3.1&nbsp;RabbitMQ Binder Properties</h3></div></div></div><p>By default, the RabbitMQ binder uses Spring Boot&#8217;s <code class="literal">ConnectionFactory</code>, and it therefore supports all Spring Boot configuration options for RabbitMQ.
(For reference, consult the <a class="link" href="http://docs.spring.io/spring-boot/docs/current/reference/htmlsingle/#common-application-properties" target="_top">Spring Boot documentation</a>.)
RabbitMQ configuration options use the <code class="literal">spring.rabbitmq</code> prefix.</p><p>In addition to Spring Boot options, the RabbitMQ binder supports the following properties:</p><div class="variablelist"><dl class="variablelist"><dt><span class="term">spring.cloud.stream.rabbit.binder.adminAddresses</span></dt><dd><p class="simpara"> A comma-separated list of RabbitMQ management plugin URLs.
Only used when <code class="literal">nodes</code> contains more than one entry.
Each entry in this list must have a corresponding entry in <code class="literal">spring.rabbitmq.addresses</code>.</p><p class="simpara">Default: empty.</p></dd><dt><span class="term">spring.cloud.stream.rabbit.binder.nodes</span></dt><dd><p class="simpara"> 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 <code class="literal">spring.rabbitmq.addresses</code>.</p><p class="simpara">Default: empty.</p></dd><dt><span class="term">spring.cloud.stream.rabbit.binder.compressionLevel</span></dt><dd><p class="simpara"> Compression level for compressed bindings.
See <code class="literal">java.util.zip.Deflater</code>.</p><p class="simpara">Default: <code class="literal">1</code> (BEST_LEVEL).</p></dd></dl></div></div><div class="section"><div class="titlepage"><div><div><h3 class="title"><a name="_rabbitmq_consumer_properties" href="#_rabbitmq_consumer_properties"></a>38.3.2&nbsp;RabbitMQ Consumer Properties</h3></div></div></div><p>The following properties are available for Rabbit consumers only and
must be prefixed with <code class="literal">spring.cloud.stream.rabbit.bindings.&lt;channelName&gt;.consumer.</code>.</p><div class="variablelist"><dl class="variablelist"><dt><span class="term">acknowledgeMode</span></dt><dd><p class="simpara">The acknowledge mode.</p><p class="simpara">Default: <code class="literal">AUTO</code>.</p></dd><dt><span class="term">autoBindDlq</span></dt><dd><p class="simpara">Whether to automatically declare the DLQ and bind it to the binder DLX.</p><p class="simpara">Default: <code class="literal">false</code>.</p></dd><dt><span class="term">bindingRoutingKey</span></dt><dd><p class="simpara">The routing key with which to bind the queue to the exchange (if <code class="literal">bindQueue</code> is <code class="literal">true</code>).
for partitioned destinations <code class="literal">-&lt;instanceIndex&gt;</code> will be appended.</p><p class="simpara">Default: <code class="literal">#</code>.</p></dd><dt><span class="term">bindQueue</span></dt><dd><p class="simpara">Whether to bind the queue to the destination exchange; set to <code class="literal">false</code> if you have set up your own infrastructure and have previously created/bound the queue.</p><p class="simpara">Default: <code class="literal">true</code>.</p></dd><dt><span class="term">deadLetterQueueName</span></dt><dd><p class="simpara">name of the DLQ</p><p class="simpara">Default: <code class="literal">prefix+destination.dlq</code></p></dd><dt><span class="term">deadLetterExchange</span></dt><dd><p class="simpara">a DLX to assign to the queue; if autoBindDlq is true</p><p class="simpara">Default: 'prefix+DLX'</p></dd><dt><span class="term">deadLetterRoutingKey</span></dt><dd><p class="simpara">a dead letter routing key to assign to the queue; if autoBindDlq is true</p><p class="simpara">Default: <code class="literal">destination</code></p></dd><dt><span class="term">declareExchange</span></dt><dd><p class="simpara">Whether to declare the exchange for the destination.</p><p class="simpara">Default: <code class="literal">true</code>.</p></dd><dt><span class="term">delayedExchange</span></dt><dd><p class="simpara">Whether to declare the exchange as a <code class="literal">Delayed Message Exchange</code> - requires the delayed message exchange plugin on the broker.
The <code class="literal">x-delayed-type</code> argument is set to the <code class="literal">exchangeType</code>.</p><p class="simpara">Default: <code class="literal">false</code>.</p></dd><dt><span class="term">dlqDeadLetterExchange</span></dt><dd><p class="simpara">if a DLQ is declared, a DLX to assign to that queue</p><p class="simpara">Default: <code class="literal">none</code></p></dd><dt><span class="term">dlqDeadLetterRoutingKey</span></dt><dd><p class="simpara">if a DLQ is declared, a dead letter routing key to assign to that queue; default none</p><p class="simpara">Default: <code class="literal">none</code></p></dd><dt><span class="term">dlqExpires</span></dt><dd><p class="simpara">how long before an unused dead letter queue is deleted (ms)</p><p class="simpara">Default: <code class="literal">no expiration</code></p></dd><dt><span class="term">dlqLazy</span></dt><dd><p class="simpara">Declare the dead letter queue with the <code class="literal">x-queue-mode=lazy</code> argument.
See <a class="link" href="https://www.rabbitmq.com/lazy-queues.html" target="_top">Lazy Queues</a>.
Consider using a policy instead of this setting because using a policy allows changing the setting without deleting the queue.</p><p class="simpara">Default: <code class="literal">false</code>.</p></dd><dt><span class="term">dlqMaxLength</span></dt><dd><p class="simpara">maximum number of messages in the dead letter queue</p><p class="simpara">Default: <code class="literal">no limit</code></p></dd><dt><span class="term">dlqMaxLengthBytes</span></dt><dd><p class="simpara">maximum number of total bytes in the dead letter queue from all messages</p><p class="simpara">Default: <code class="literal">no limit</code></p></dd><dt><span class="term">dlqMaxPriority</span></dt><dd><p class="simpara">maximum priority of messages in the dead letter queue (0-255)</p><p class="simpara">Default: <code class="literal">none</code></p></dd><dt><span class="term">dlqTtl</span></dt><dd><p class="simpara">default time to live to apply to the dead letter queue when declared (ms)</p><p class="simpara">Default: <code class="literal">no limit</code></p></dd><dt><span class="term">durableSubscription</span></dt><dd><p class="simpara"> Whether subscription should be durable.
Only effective if <code class="literal">group</code> is also set.</p><p class="simpara">Default: <code class="literal">true</code>.</p></dd><dt><span class="term">exchangeAutoDelete</span></dt><dd><p class="simpara">If <code class="literal">declareExchange</code> is true, whether the exchange should be auto-delete (removed after the last queue is removed).</p><p class="simpara">Default: <code class="literal">true</code>.</p></dd><dt><span class="term">exchangeDurable</span></dt><dd><p class="simpara">If <code class="literal">declareExchange</code> is true, whether the exchange should be durable (survives broker restart).</p><p class="simpara">Default: <code class="literal">true</code>.</p></dd><dt><span class="term">exchangeType</span></dt><dd><p class="simpara">The exchange type; <code class="literal">direct</code>, <code class="literal">fanout</code> or <code class="literal">topic</code> for non-partitioned destinations; <code class="literal">direct</code> or <code class="literal">topic</code> for partitioned destinations.</p><p class="simpara">Default: <code class="literal">topic</code>.</p></dd><dt><span class="term">exclusive</span></dt><dd><p class="simpara">Create an exclusive consumer; concurrency should be 1 when this is <code class="literal">true</code>; often used when strict ordering is required but enabling a hot standby instance to take over after a failure.
See <code class="literal">recoveryInterval</code>, which controls how often a standby instance will attempt to consume.</p><p class="simpara">Default: <code class="literal">false</code>.</p></dd><dt><span class="term">expires</span></dt><dd><p class="simpara">how long before an unused queue is deleted (ms)</p><p class="simpara">Default: <code class="literal">no expiration</code></p></dd><dt><span class="term">headerPatterns</span></dt><dd><p class="simpara">Patterns for headers to be mapped from inbound messages.</p><p class="simpara">Default: <code class="literal">['*']</code> (all headers).</p></dd><dt><span class="term">lazy</span></dt><dd><p class="simpara">Declare the queue with the <code class="literal">x-queue-mode=lazy</code> argument.
See <a class="link" href="https://www.rabbitmq.com/lazy-queues.html" target="_top">Lazy Queues</a>.
Consider using a policy instead of this setting because using a policy allows changing the setting without deleting the queue.</p><p class="simpara">Default: <code class="literal">false</code>.</p></dd><dt><span class="term">maxConcurrency</span></dt><dd><p class="simpara">the maximum number of consumers</p><p class="simpara">Default: <code class="literal">1</code>.</p></dd><dt><span class="term">maxLength</span></dt><dd><p class="simpara">maximum number of messages in the queue</p><p class="simpara">Default: <code class="literal">no limit</code></p></dd><dt><span class="term">maxLengthBytes</span></dt><dd><p class="simpara">maximum number of total bytes in the queue from all messages</p><p class="simpara">Default: <code class="literal">no limit</code></p></dd><dt><span class="term">maxPriority</span></dt><dd>maximum priority of messages in the queue (0-255)</dd><dt><span class="term">Default</span></dt><dd><code class="literal">none</code></dd><dt><span class="term">prefetch</span></dt><dd><p class="simpara">Prefetch count.</p><p class="simpara">Default: <code class="literal">1</code>.</p></dd><dt><span class="term">prefix</span></dt><dd><p class="simpara">A prefix to be added to the name of the <code class="literal">destination</code> and queues.</p><p class="simpara">Default: "".</p></dd><dt><span class="term">recoveryInterval</span></dt><dd><p class="simpara">The interval between connection recovery attempts, in milliseconds.</p><p class="simpara">Default: <code class="literal">5000</code>.</p></dd><dt><span class="term">requeueRejected</span></dt><dd><p class="simpara">Whether delivery failures should be requeued when retry is disabled or republishToDlq is false.</p><p class="simpara">Default: <code class="literal">false</code>.</p></dd><dt><span class="term">republishDeliveryMode</span></dt><dd><p class="simpara">When <code class="literal">republishToDlq</code> is <code class="literal">true</code>, specify the delivery mode of the republished message.</p><p class="simpara">Default: <code class="literal">DeliveryMode.PERSISTENT</code></p></dd><dt><span class="term">republishToDlq</span></dt><dd><p class="simpara"> 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 <code class="literal">true</code>, 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.</p><p class="simpara">Default: false</p></dd><dt><span class="term">transacted</span></dt><dd><p class="simpara">Whether to use transacted channels.</p><p class="simpara">Default: <code class="literal">false</code>.</p></dd><dt><span class="term">ttl</span></dt><dd><p class="simpara">default time to live to apply to the queue when declared (ms)</p><p class="simpara">Default: <code class="literal">no limit</code></p></dd><dt><span class="term">txSize</span></dt><dd><p class="simpara">The number of deliveries between acks.</p><p class="simpara">Default: <code class="literal">1</code>.</p></dd></dl></div></div><div class="section"><div class="titlepage"><div><div><h3 class="title"><a name="_rabbit_producer_properties" href="#_rabbit_producer_properties"></a>38.3.3&nbsp;Rabbit Producer Properties</h3></div></div></div><p>The following properties are available for Rabbit producers only and
must be prefixed with <code class="literal">spring.cloud.stream.rabbit.bindings.&lt;channelName&gt;.producer.</code>.</p><div class="variablelist"><dl class="variablelist"><dt><span class="term">autoBindDlq</span></dt><dd><p class="simpara">Whether to automatically declare the DLQ and bind it to the binder DLX.</p><p class="simpara">Default: <code class="literal">false</code>.</p></dd><dt><span class="term">batchingEnabled</span></dt><dd><p class="simpara">Whether to enable message batching by producers.</p><p class="simpara">Default: <code class="literal">false</code>.</p></dd><dt><span class="term">batchSize</span></dt><dd><p class="simpara">The number of messages to buffer when batching is enabled.</p><p class="simpara">Default: <code class="literal">100</code>.</p></dd><dt><span class="term">batchBufferLimit</span></dt><dd>Default: <code class="literal">10000</code>.</dd><dt><span class="term">batchTimeout</span></dt><dd>Default: <code class="literal">5000</code>.</dd><dt><span class="term">bindingRoutingKey</span></dt><dd><p class="simpara">The routing key with which to bind the queue to the exchange (if <code class="literal">bindQueue</code> is <code class="literal">true</code>).
Only applies to non-partitioned destinations.
Only applies if <code class="literal">requiredGroups</code> are provided and then only to those groups.</p><p class="simpara">Default: <code class="literal">#</code>.</p></dd><dt><span class="term">bindQueue</span></dt><dd><p class="simpara">Whether to bind the queue to the destination exchange; set to <code class="literal">false</code> if you have set up your own infrastructure and have previously created/bound the queue.
Only applies if <code class="literal">requiredGroups</code> are provided and then only to those groups.</p><p class="simpara">Default: <code class="literal">true</code>.</p></dd><dt><span class="term">compress</span></dt><dd><p class="simpara">Whether data should be compressed when sent.</p><p class="simpara">Default: <code class="literal">false</code>.</p></dd><dt><span class="term">deadLetterQueueName</span></dt><dd><p class="simpara">name of the DLQ
Only applies if <code class="literal">requiredGroups</code> are provided and then only to those groups.</p><p class="simpara">Default: <code class="literal">prefix+destination.dlq</code></p></dd><dt><span class="term">deadLetterExchange</span></dt><dd><p class="simpara">a DLX to assign to the queue; if autoBindDlq is true
Only applies if <code class="literal">requiredGroups</code> are provided and then only to those groups.</p><p class="simpara">Default: 'prefix+DLX'</p></dd><dt><span class="term">deadLetterRoutingKey</span></dt><dd><p class="simpara">a dead letter routing key to assign to the queue; if autoBindDlq is true
Only applies if <code class="literal">requiredGroups</code> are provided and then only to those groups.</p><p class="simpara">Default: <code class="literal">destination</code></p></dd><dt><span class="term">declareExchange</span></dt><dd><p class="simpara">Whether to declare the exchange for the destination.</p><p class="simpara">Default: <code class="literal">true</code>.</p></dd><dt><span class="term">delay</span></dt><dd><p class="simpara">A SpEL expression to evaluate the delay to apply to the message (<code class="literal">x-delay</code> header) - has no effect if the exchange is not a delayed message exchange.</p><p class="simpara">Default: No <code class="literal">x-delay</code> header is set.</p></dd><dt><span class="term">delayedExchange</span></dt><dd><p class="simpara">Whether to declare the exchange as a <code class="literal">Delayed Message Exchange</code> - requires the delayed message exchange plugin on the broker.
The <code class="literal">x-delayed-type</code> argument is set to the <code class="literal">exchangeType</code>.</p><p class="simpara">Default: <code class="literal">false</code>.</p></dd><dt><span class="term">deliveryMode</span></dt><dd><p class="simpara">Delivery mode.</p><p class="simpara">Default: <code class="literal">PERSISTENT</code>.</p></dd><dt><span class="term">dlqDeadLetterExchange</span></dt><dd><p class="simpara">if a DLQ is declared, a DLX to assign to that queue
Only applies if <code class="literal">requiredGroups</code> are provided and then only to those groups.</p><p class="simpara">Default: <code class="literal">none</code></p></dd><dt><span class="term">dlqDeadLetterRoutingKey</span></dt><dd><p class="simpara">if a DLQ is declared, a dead letter routing key to assign to that queue; default none
Only applies if <code class="literal">requiredGroups</code> are provided and then only to those groups.</p><p class="simpara">Default: <code class="literal">none</code></p></dd><dt><span class="term">dlqExpires</span></dt><dd><p class="simpara">how long before an unused dead letter queue is deleted (ms)
Only applies if <code class="literal">requiredGroups</code> are provided and then only to those groups.</p><p class="simpara">Default: <code class="literal">no expiration</code></p></dd><dt><span class="term">dlqLazy</span></dt><dd>Declare the dead letter queue with the <code class="literal">x-queue-mode=lazy</code> argument.
See <a class="link" href="https://www.rabbitmq.com/lazy-queues.html" target="_top">Lazy Queues</a>.
Consider using a policy instead of this setting because using a policy allows changing the setting without deleting the queue.
Only applies if <code class="literal">requiredGroups</code> are provided and then only to those groups.</dd><dt><span class="term">dlqMaxLength</span></dt><dd><p class="simpara">maximum number of messages in the dead letter queue
Only applies if <code class="literal">requiredGroups</code> are provided and then only to those groups.</p><p class="simpara">Default: <code class="literal">no limit</code></p></dd><dt><span class="term">dlqMaxLengthBytes</span></dt><dd><p class="simpara">maximum number of total bytes in the dead letter queue from all messages
Only applies if <code class="literal">requiredGroups</code> are provided and then only to those groups.</p><p class="simpara">Default: <code class="literal">no limit</code></p></dd><dt><span class="term">dlqMaxPriority</span></dt><dd><p class="simpara">maximum priority of messages in the dead letter queue (0-255)
Only applies if <code class="literal">requiredGroups</code> are provided and then only to those groups.</p><p class="simpara">Default: <code class="literal">none</code></p></dd><dt><span class="term">dlqTtl</span></dt><dd><p class="simpara">default time to live to apply to the dead letter queue when declared (ms)
Only applies if <code class="literal">requiredGroups</code> are provided and then only to those groups.</p><p class="simpara">Default: <code class="literal">no limit</code></p></dd><dt><span class="term">exchangeAutoDelete</span></dt><dd><p class="simpara">If <code class="literal">declareExchange</code> is true, whether the exchange should be auto-delete (removed after the last queue is removed).</p><p class="simpara">Default: <code class="literal">true</code>.</p></dd><dt><span class="term">exchangeDurable</span></dt><dd><p class="simpara">If <code class="literal">declareExchange</code> is true, whether the exchange should be durable (survives broker restart).</p><p class="simpara">Default: <code class="literal">true</code>.</p></dd><dt><span class="term">exchangeType</span></dt><dd><p class="simpara">The exchange type; <code class="literal">direct</code>, <code class="literal">fanout</code> or <code class="literal">topic</code> for non-partitioned destinations; <code class="literal">direct</code> or <code class="literal">topic</code> for partitioned destinations.</p><p class="simpara">Default: <code class="literal">topic</code>.</p></dd><dt><span class="term">expires</span></dt><dd><p class="simpara">how long before an unused queue is deleted (ms)
Only applies if <code class="literal">requiredGroups</code> are provided and then only to those groups.</p><p class="simpara">Default: <code class="literal">no expiration</code></p></dd><dt><span class="term">headerPatterns</span></dt><dd><p class="simpara">Patterns for headers to be mapped to outbound messages.</p><p class="simpara">Default: <code class="literal">['*']</code> (all headers).</p></dd><dt><span class="term">lazy</span></dt><dd><p class="simpara">Declare the queue with the <code class="literal">x-queue-mode=lazy</code> argument.
See <a class="link" href="https://www.rabbitmq.com/lazy-queues.html" target="_top">Lazy Queues</a>.
Consider using a policy instead of this setting because using a policy allows changing the setting without deleting the queue.
Only applies if <code class="literal">requiredGroups</code> are provided and then only to those groups.</p><p class="simpara">Default: <code class="literal">false</code>.</p></dd><dt><span class="term">maxLength</span></dt><dd><p class="simpara">maximum number of messages in the queue
Only applies if <code class="literal">requiredGroups</code> are provided and then only to those groups.</p><p class="simpara">Default: <code class="literal">no limit</code></p></dd><dt><span class="term">maxLengthBytes</span></dt><dd><p class="simpara">maximum number of total bytes in the queue from all messages
Only applies if <code class="literal">requiredGroups</code> are provided and then only to those groups.</p><p class="simpara">Default: <code class="literal">no limit</code></p></dd><dt><span class="term">maxPriority</span></dt><dd>maximum priority of messages in the queue (0-255)
Only applies if <code class="literal">requiredGroups</code> are provided and then only to those groups.</dd><dt><span class="term">Default</span></dt><dd><code class="literal">none</code></dd><dt><span class="term">prefix</span></dt><dd><p class="simpara">A prefix to be added to the name of the <code class="literal">destination</code> exchange.</p><p class="simpara">Default: "".</p></dd><dt><span class="term">routingKeyExpression</span></dt><dd><p class="simpara">A SpEL expression to determine the routing key to use when publishing messages.
For a fixed routing key, use a literal expression, e.g. <code class="literal">routingKeyExpression='my.routingKey'</code> in a properties file, or <code class="literal">routingKeyExpression: '''my.routingKey'''</code> in a YAML file.</p><p class="simpara">Default: <code class="literal">destination</code> or <code class="literal">destination-&lt;partition&gt;</code> for partitioned destinations.</p></dd><dt><span class="term">transacted</span></dt><dd><p class="simpara">Whether to use transacted channels.</p><p class="simpara">Default: <code class="literal">false</code>.</p></dd><dt><span class="term">ttl</span></dt><dd><p class="simpara">default time to live to apply to the queue when declared (ms)
Only applies if <code class="literal">requiredGroups</code> are provided and then only to those groups.</p><p class="simpara">Default: <code class="literal">no limit</code></p></dd></dl></div><div class="note" style="margin-left: 0.5in; margin-right: 0.5in;"><table border="0" summary="Note"><tr><td rowspan="2" align="center" valign="top" width="25"><img alt="[Note]" src="images/note.png"></td><th align="left">Note</th></tr><tr><td align="left" valign="top"><p>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, that do not normally support headers).</p></td></tr></table></div></div></div><div class="section"><div class="titlepage"><div><div><h2 class="title" style="clear: both"><a name="_retry_with_the_rabbitmq_binder" href="#_retry_with_the_rabbitmq_binder"></a>38.4&nbsp;Retry With the RabbitMQ Binder</h2></div></div></div><div class="section"><div class="titlepage"><div><div><h3 class="title"><a name="_overview" href="#_overview"></a>38.4.1&nbsp;Overview</h3></div></div></div><p>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 <a class="xref" href="multi__rabbitmq_binder.html#rabbit-binder-properties" title="38.3.1&nbsp;RabbitMQ Binder Properties">Section&nbsp;38.3.1, &#8220;RabbitMQ Binder Properties&#8221;</a> for more information about the properties discussed here.
Example configuration to enable this feature:</p><div class="itemizedlist"><ul class="itemizedlist" style="list-style-type: disc; "><li class="listitem">Set <code class="literal">autoBindDlq</code> to <code class="literal">true</code> - the binder will create a DLQ; you can optionally specify a name in <code class="literal">deadLetterQueueName</code></li><li class="listitem">Set <code class="literal">dlqTtl</code> to the back off time you want to wait between redeliveries</li><li class="listitem">Set the <code class="literal">dlqDeadLetterExchange</code> to the default exchange - expired messages from the DLQ will be routed to the original queue since the default <code class="literal">deadLetterRoutingKey</code> is the queue name (<code class="literal">destination.group</code>)</li></ul></div><p>To force a message to be dead-lettered, either throw an <code class="literal">AmqpRejectAndDontRequeueException</code>, or set <code class="literal">requeueRejected</code> to <code class="literal">true</code> and throw any exception.</p><p>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 <code class="literal">x-death</code> header which allows you to determine how many cycles have occurred.</p><p>To acknowledge a message after giving up, throw an <code class="literal">ImmediateAcknowledgeAmqpException</code>.</p></div><div class="section"><div class="titlepage"><div><div><h3 class="title"><a name="_putting_it_all_together" href="#_putting_it_all_together"></a>38.4.2&nbsp;Putting it All Together</h3></div></div></div><pre class="screen">---
spring.cloud.stream.bindings.input.destination=myDestination
spring.cloud.stream.bindings.input.group=consumerGroup
#disable binder retries
spring.cloud.stream.bindings.input.consumer.max-attempts=1
#dlx/dlq setup
spring.cloud.stream.rabbit.bindings.input.consumer.auto-bind-dlq=true
spring.cloud.stream.rabbit.bindings.input.consumer.dlq-ttl=5000
spring.cloud.stream.rabbit.bindings.input.consumer.dlq-dead-letter-exchange=
---</pre><p>This configuration creates an exchange <code class="literal">myDestination</code> with queue <code class="literal">myDestination.consumerGroup</code> bound to a topic exchange with a wildcard routing key <code class="literal">#</code>.
It creates a DLQ bound to a direct exchange <code class="literal">DLX</code> with routing key <code class="literal">myDestination.consumerGroup</code>.
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.</p><p><b>Spring Boot application.&nbsp;</b>
</p><pre class="programlisting"><em><span class="hl-annotation" style="color: gray">@SpringBootApplication</span></em>
<em><span class="hl-annotation" style="color: gray">@EnableBinding(Sink.class)</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> XDeathApplication {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">void</span> main(String[] args) {
SpringApplication.run(XDeathApplication.<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span>, args);
}
<em><span class="hl-annotation" style="color: gray">@StreamListener(Sink.INPUT)</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">void</span> listen(String in, <em><span class="hl-annotation" style="color: gray">@Header(name = "x-death", required = false)</span></em> Map&lt;?,?&gt; death) {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">if</span> (death != null &amp;&amp; death.get(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"count"</span>).equals(<span class="hl-number">3L</span>)) {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-comment">// giving up - don't send to DLX</span>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">throw</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">new</span> ImmediateAcknowledgeAmqpException(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"Failed after 4 attempts"</span>);
}
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">throw</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">new</span> AmqpRejectAndDontRequeueException(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"failed"</span>);
}
}</pre><p>
</p><p>Notice that the count property in the <code class="literal">x-death</code> header is a <code class="literal">Long</code>.</p></div></div><div class="section"><div class="titlepage"><div><div><h2 class="title" style="clear: both"><a name="rabbit-error-channels" href="#rabbit-error-channels"></a>38.5&nbsp;Error Channels</h2></div></div></div><p>Starting with <span class="emphasis"><em>version 1.3</em></span>, 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 <a class="xref" href="multi__programming_model.html#binder-error-channels" title="Message Channel Binders and Error Channels">the section called &#8220;Message Channel Binders and Error Channels&#8221;</a> for more information.</p><p>With rabbitmq, there are two types of send failures:</p><div class="itemizedlist"><ul class="itemizedlist" style="list-style-type: disc; "><li class="listitem">returned messages</li><li class="listitem">negatively acknowledged <a class="link" href="https://www.rabbitmq.com/confirms.html" target="_top">Publisher Confirms</a></li></ul></div><p>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.".</p><p>As well as enabling producer error channels as described in <a class="xref" href="multi__programming_model.html#binder-error-channels" title="Message Channel Binders and Error Channels">the section called &#8220;Message Channel Binders and Error Channels&#8221;</a>, the RabbitMQ binder will only send messages to the channels if the connection factory is appropriately configured:</p><div class="itemizedlist"><ul class="itemizedlist" style="list-style-type: disc; "><li class="listitem"><code class="literal">ccf.setPublisherConfirms(true);</code></li><li class="listitem"><code class="literal">ccf.setPublisherReturns(true);</code></li></ul></div><p>When using spring boot configuration for the connection factory, set properties:</p><div class="itemizedlist"><ul class="itemizedlist" style="list-style-type: disc; "><li class="listitem"><code class="literal">spring.rabbitmq.publisher-confirms</code></li><li class="listitem"><code class="literal">spring.rabbitmq.publisher-returns</code></li></ul></div><p>The payload of the <code class="literal">ErrorMessage</code> for a returned message is a <code class="literal">ReturnedAmqpMessageException</code> with properties:</p><div class="itemizedlist"><ul class="itemizedlist" style="list-style-type: disc; "><li class="listitem"><code class="literal">failedMessage</code> - the spring-messaging <code class="literal">Message&lt;?&gt;</code> that failed to be sent.</li><li class="listitem"><code class="literal">amqpMessage</code> - the raw spring-amqp <code class="literal">Message</code></li><li class="listitem"><code class="literal">replyCode</code> - an integer value indicating the reason for the failure (e.g. 312 - No route)</li><li class="listitem"><code class="literal">replyText</code> - a text value indicating the reason for the failure e.g. <code class="literal">NO_ROUTE</code>.</li><li class="listitem"><code class="literal">exchange</code> - the exchange to which the message was published.</li><li class="listitem"><code class="literal">routingKey</code> - the routing key used when the message was published.</li></ul></div><p>For negatively acknowledged confirms, the payload is a <code class="literal">NackedAmqpMessageException</code> with properties:</p><div class="itemizedlist"><ul class="itemizedlist" style="list-style-type: disc; "><li class="listitem"><code class="literal">failedMessage</code> - the spring-messaging <code class="literal">Message&lt;?&gt;</code> that failed to be sent.</li><li class="listitem"><code class="literal">nackReason</code> - a reason (if available; you may need to examine the broker logs for more information).</li></ul></div><p>There is no automatic handling of these exceptions (such as sending to a <a class="link" href="multi__rabbitmq_binder.html#rabbit-dlq-processing" title="38.6&nbsp;Dead-Letter Queue Processing">Dead-Letter queue</a>); you can consume these exceptions with your own Spring Integration flow.</p></div><div class="section"><div class="titlepage"><div><div><h2 class="title" style="clear: both"><a name="rabbit-dlq-processing" href="#rabbit-dlq-processing"></a>38.6&nbsp;Dead-Letter Queue Processing</h2></div></div></div><p>Because it can&#8217;t be anticipated 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 <code class="literal">spring-boot</code> 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 <a class="link" href="https://www.rabbitmq.com/blog/2015/04/16/scheduling-messages-with-rabbitmq/" target="_top">RabbitMQ Delayed Message Exchange</a> to introduce a delay to the requeued message.
In this example, the delay increases for each attempt.
These examples use a <code class="literal">@RabbitListener</code> to receive messages from the DLQ, you could also use <code class="literal">RabbitTemplate.receive()</code> in a batch process.</p><p>The examples assume the original destination is <code class="literal">so8400in</code> and the consumer group is <code class="literal">so8400</code>.</p><div class="section"><div class="titlepage"><div><div><h3 class="title"><a name="_non_partitioned_destinations" href="#_non_partitioned_destinations"></a>38.6.1&nbsp;Non-Partitioned Destinations</h3></div></div></div><p>The first two examples are when the destination is <span class="strong"><strong>not</strong></span> partitioned.</p><pre class="programlisting"><em><span class="hl-annotation" style="color: gray">@SpringBootApplication</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> ReRouteDlqApplication {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">final</span> String ORIGINAL_QUEUE = <span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"so8400in.so8400"</span>;
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">final</span> String DLQ = ORIGINAL_QUEUE + <span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">".dlq"</span>;
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">final</span> String PARKING_LOT = ORIGINAL_QUEUE + <span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">".parkingLot"</span>;
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">final</span> String X_RETRIES_HEADER = <span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"x-retries"</span>;
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">void</span> main(String[] args) <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">throws</span> Exception {
ConfigurableApplicationContext context = SpringApplication.run(ReRouteDlqApplication.<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span>, args);
System.out.println(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"Hit enter to terminate"</span>);
System.in.read();
context.close();
}
<em><span class="hl-annotation" style="color: gray">@Autowired</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> RabbitTemplate rabbitTemplate;
<em><span class="hl-annotation" style="color: gray">@RabbitListener(queues = DLQ)</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">void</span> rePublish(Message failedMessage) {
Integer retriesHeader = (Integer) failedMessage.getMessageProperties().getHeaders().get(X_RETRIES_HEADER);
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">if</span> (retriesHeader == null) {
retriesHeader = Integer.valueOf(<span class="hl-number">0</span>);
}
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">if</span> (retriesHeader &lt; <span class="hl-number">3</span>) {
failedMessage.getMessageProperties().getHeaders().put(X_RETRIES_HEADER, retriesHeader + <span class="hl-number">1</span>);
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">this</span>.rabbitTemplate.send(ORIGINAL_QUEUE, failedMessage);
}
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">else</span> {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">this</span>.rabbitTemplate.send(PARKING_LOT, failedMessage);
}
}
<em><span class="hl-annotation" style="color: gray">@Bean</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> Queue parkingLot() {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">return</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">new</span> Queue(PARKING_LOT);
}
}</pre><pre class="programlisting"><em><span class="hl-annotation" style="color: gray">@SpringBootApplication</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> ReRouteDlqApplication {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">final</span> String ORIGINAL_QUEUE = <span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"so8400in.so8400"</span>;
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">final</span> String DLQ = ORIGINAL_QUEUE + <span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">".dlq"</span>;
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">final</span> String PARKING_LOT = ORIGINAL_QUEUE + <span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">".parkingLot"</span>;
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">final</span> String X_RETRIES_HEADER = <span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"x-retries"</span>;
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">final</span> String DELAY_EXCHANGE = <span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"dlqReRouter"</span>;
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">void</span> main(String[] args) <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">throws</span> Exception {
ConfigurableApplicationContext context = SpringApplication.run(ReRouteDlqApplication.<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span>, args);
System.out.println(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"Hit enter to terminate"</span>);
System.in.read();
context.close();
}
<em><span class="hl-annotation" style="color: gray">@Autowired</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> RabbitTemplate rabbitTemplate;
<em><span class="hl-annotation" style="color: gray">@RabbitListener(queues = DLQ)</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">void</span> rePublish(Message failedMessage) {
Map&lt;String, Object&gt; headers = failedMessage.getMessageProperties().getHeaders();
Integer retriesHeader = (Integer) headers.get(X_RETRIES_HEADER);
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">if</span> (retriesHeader == null) {
retriesHeader = Integer.valueOf(<span class="hl-number">0</span>);
}
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">if</span> (retriesHeader &lt; <span class="hl-number">3</span>) {
headers.put(X_RETRIES_HEADER, retriesHeader + <span class="hl-number">1</span>);
headers.put(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"x-delay"</span>, <span class="hl-number">5000</span> * retriesHeader);
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">this</span>.rabbitTemplate.send(DELAY_EXCHANGE, ORIGINAL_QUEUE, failedMessage);
}
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">else</span> {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">this</span>.rabbitTemplate.send(PARKING_LOT, failedMessage);
}
}
<em><span class="hl-annotation" style="color: gray">@Bean</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> DirectExchange delayExchange() {
DirectExchange exchange = <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">new</span> DirectExchange(DELAY_EXCHANGE);
exchange.setDelayed(true);
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">return</span> exchange;
}
<em><span class="hl-annotation" style="color: gray">@Bean</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> Binding bindOriginalToDelay() {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">return</span> BindingBuilder.bind(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">new</span> Queue(ORIGINAL_QUEUE)).to(delayExchange()).with(ORIGINAL_QUEUE);
}
<em><span class="hl-annotation" style="color: gray">@Bean</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> Queue parkingLot() {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">return</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">new</span> Queue(PARKING_LOT);
}
}</pre></div><div class="section"><div class="titlepage"><div><div><h3 class="title"><a name="_partitioned_destinations" href="#_partitioned_destinations"></a>38.6.2&nbsp;Partitioned Destinations</h3></div></div></div><p>With partitioned destinations, there is one DLQ for all partitions and we determine the original queue from the headers.</p><div class="section"><div class="titlepage"><div><div><h4 class="title"><a name="_republishtodlq_false" href="#_republishtodlq_false"></a>republishToDlq=false</h4></div></div></div><p>When <code class="literal">republishToDlq</code> is <code class="literal">false</code>, RabbitMQ publishes the message to the DLX/DLQ with an <code class="literal">x-death</code> header containing information about the original destination.</p><pre class="programlisting"><em><span class="hl-annotation" style="color: gray">@SpringBootApplication</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> ReRouteDlqApplication {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">final</span> String ORIGINAL_QUEUE = <span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"so8400in.so8400"</span>;
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">final</span> String DLQ = ORIGINAL_QUEUE + <span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">".dlq"</span>;
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">final</span> String PARKING_LOT = ORIGINAL_QUEUE + <span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">".parkingLot"</span>;
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">final</span> String X_DEATH_HEADER = <span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"x-death"</span>;
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">final</span> String X_RETRIES_HEADER = <span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"x-retries"</span>;
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">void</span> main(String[] args) <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">throws</span> Exception {
ConfigurableApplicationContext context = SpringApplication.run(ReRouteDlqApplication.<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span>, args);
System.out.println(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"Hit enter to terminate"</span>);
System.in.read();
context.close();
}
<em><span class="hl-annotation" style="color: gray">@Autowired</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> RabbitTemplate rabbitTemplate;
<em><span class="hl-annotation" style="color: gray">@SuppressWarnings("unchecked")</span></em>
<em><span class="hl-annotation" style="color: gray">@RabbitListener(queues = DLQ)</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">void</span> rePublish(Message failedMessage) {
Map&lt;String, Object&gt; headers = failedMessage.getMessageProperties().getHeaders();
Integer retriesHeader = (Integer) headers.get(X_RETRIES_HEADER);
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">if</span> (retriesHeader == null) {
retriesHeader = Integer.valueOf(<span class="hl-number">0</span>);
}
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">if</span> (retriesHeader &lt; <span class="hl-number">3</span>) {
headers.put(X_RETRIES_HEADER, retriesHeader + <span class="hl-number">1</span>);
List&lt;Map&lt;String, ?&gt;&gt; xDeath = (List&lt;Map&lt;String, ?&gt;&gt;) headers.get(X_DEATH_HEADER);
String exchange = (String) xDeath.get(<span class="hl-number">0</span>).get(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"exchange"</span>);
List&lt;String&gt; routingKeys = (List&lt;String&gt;) xDeath.get(<span class="hl-number">0</span>).get(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"routing-keys"</span>);
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">this</span>.rabbitTemplate.send(exchange, routingKeys.get(<span class="hl-number">0</span>), failedMessage);
}
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">else</span> {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">this</span>.rabbitTemplate.send(PARKING_LOT, failedMessage);
}
}
<em><span class="hl-annotation" style="color: gray">@Bean</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> Queue parkingLot() {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">return</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">new</span> Queue(PARKING_LOT);
}
}</pre></div><div class="section"><div class="titlepage"><div><div><h4 class="title"><a name="_republishtodlq_true" href="#_republishtodlq_true"></a>republishToDlq=true</h4></div></div></div><p>When <code class="literal">republishToDlq</code> is <code class="literal">true</code>, the republishing recoverer adds the original exchange and routing key to headers.</p><pre class="programlisting"><em><span class="hl-annotation" style="color: gray">@SpringBootApplication</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span> ReRouteDlqApplication {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">final</span> String ORIGINAL_QUEUE = <span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"so8400in.so8400"</span>;
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">final</span> String DLQ = ORIGINAL_QUEUE + <span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">".dlq"</span>;
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">final</span> String PARKING_LOT = ORIGINAL_QUEUE + <span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">".parkingLot"</span>;
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">final</span> String X_RETRIES_HEADER = <span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"x-retries"</span>;
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">final</span> String X_ORIGINAL_EXCHANGE_HEADER = RepublishMessageRecoverer.X_ORIGINAL_EXCHANGE;
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">final</span> String X_ORIGINAL_ROUTING_KEY_HEADER = RepublishMessageRecoverer.X_ORIGINAL_ROUTING_KEY;
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">static</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">void</span> main(String[] args) <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">throws</span> Exception {
ConfigurableApplicationContext context = SpringApplication.run(ReRouteDlqApplication.<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">class</span>, args);
System.out.println(<span xmlns:d="http://docbook.org/ns/docbook" class="hl-string">"Hit enter to terminate"</span>);
System.in.read();
context.close();
}
<em><span class="hl-annotation" style="color: gray">@Autowired</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">private</span> RabbitTemplate rabbitTemplate;
<em><span class="hl-annotation" style="color: gray">@RabbitListener(queues = DLQ)</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">void</span> rePublish(Message failedMessage) {
Map&lt;String, Object&gt; headers = failedMessage.getMessageProperties().getHeaders();
Integer retriesHeader = (Integer) headers.get(X_RETRIES_HEADER);
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">if</span> (retriesHeader == null) {
retriesHeader = Integer.valueOf(<span class="hl-number">0</span>);
}
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">if</span> (retriesHeader &lt; <span class="hl-number">3</span>) {
headers.put(X_RETRIES_HEADER, retriesHeader + <span class="hl-number">1</span>);
String exchange = (String) headers.get(X_ORIGINAL_EXCHANGE_HEADER);
String originalRoutingKey = (String) headers.get(X_ORIGINAL_ROUTING_KEY_HEADER);
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">this</span>.rabbitTemplate.send(exchange, originalRoutingKey, failedMessage);
}
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">else</span> {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">this</span>.rabbitTemplate.send(PARKING_LOT, failedMessage);
}
}
<em><span class="hl-annotation" style="color: gray">@Bean</span></em>
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">public</span> Queue parkingLot() {
<span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">return</span> <span xmlns:d="http://docbook.org/ns/docbook" class="hl-keyword">new</span> Queue(PARKING_LOT);
}
}</pre></div></div></div></div><div class="navfooter"><hr><table width="100%" summary="Navigation footer"><tr><td width="40%" align="left"><a accesskey="p" href="multi__apache_kafka_binder.html">Prev</a>&nbsp;</td><td width="20%" align="center"><a accesskey="u" href="multi__binder_implementations.html">Up</a></td><td width="40%" align="right">&nbsp;<a accesskey="n" href="multi__spring_cloud_bus.html">Next</a></td></tr><tr><td width="40%" align="left" valign="top">37.&nbsp;Apache Kafka Binder&nbsp;</td><td width="20%" align="center"><a accesskey="h" href="multi_spring-cloud.html">Home</a></td><td width="40%" align="right" valign="top">&nbsp;Part&nbsp;VI.&nbsp;Spring Cloud Bus</td></tr></table></div></body></html>
Reference in New Issue View Git Blame Copy Permalink