root
/
spring-boot
mirror of https://github.com/spring-projects/spring-boot.git
1
0
Fork 0
Code Issues Actions 15 Packages Projects Releases Wiki Activity

Merge pull request #5341 from garyrussell/GH-5340 

* pr/5341:
  Polish contribution
  Add Retry Config: Template and Listener
This commit is contained in:
Stephane Nicoll 2016-03-17 13:26:22 +01:00
parent f4bb9e3ba7 25f00b9bb8
commit 4ebba984f0
6 changed files with 285 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitAutoConfiguration.java
View File

@ -29,6 +29,8 @@ import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties.Retry;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties.Template;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
@ -36,6 +38,9 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
/**
* {@link EnableAutoConfiguration Auto-configuration} for {@link RabbitTemplate}.
@ -94,12 +99,32 @@ public class RabbitAutoConfiguration {
@Bean
@ConditionalOnMissingBean(RabbitTemplate.class)
public RabbitTemplate rabbitTemplate() {
public RabbitTemplate rabbitTemplate(RabbitProperties config) {
RabbitTemplate rabbitTemplate = new RabbitTemplate(this.connectionFactory);
MessageConverter messageConverter = this.messageConverter.getIfUnique();
if (messageConverter != null) {
rabbitTemplate.setMessageConverter(messageConverter);
}
Template template = config.getTemplate();
Retry retry = template.getRetry();
if (retry.isEnabled()) {
RetryTemplate retryTemplate = new RetryTemplate();
SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
retryPolicy.setMaxAttempts(retry.getMaxAttempts());
retryTemplate.setRetryPolicy(retryPolicy);
ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
backOffPolicy.setInitialInterval(retry.getInitialInterval());
backOffPolicy.setMultiplier(retry.getMultiplier());
backOffPolicy.setMaxInterval(retry.getMaxInterval());
retryTemplate.setBackOffPolicy(backOffPolicy);
rabbitTemplate.setRetryTemplate(retryTemplate);
}
if (template.getReceiveTimeout() != null) {
rabbitTemplate.setReceiveTimeout(template.getReceiveTimeout());
}
if (template.getReplyTimeout() != null) {
rabbitTemplate.setReplyTimeout(template.getReplyTimeout());
}
return rabbitTemplate;
}

155
spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitProperties.java
View File

@ -22,6 +22,7 @@ import java.util.Set;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.CacheMode;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.NestedConfigurationProperty;
import org.springframework.util.StringUtils;
/**
@ -87,6 +88,8 @@ public class RabbitProperties {
*/
private final Listener listener = new Listener();
private final Template template = new Template();
public String getHost() {
if (this.addresses == null) {
return this.host;
@ -201,6 +204,10 @@ public class RabbitProperties {
return this.listener;
}
public Template getTemplate() {
return this.template;
}
public static class Ssl {
/**
@ -382,6 +389,17 @@ public class RabbitProperties {
*/
private Integer transactionSize;
/**
* Whether rejected deliveries are requeued by default; default true.
*/
private Boolean defaultRequeueRejected;
/**
* Optional properties for a retry interceptor.
*/
@NestedConfigurationProperty
private final ListenerRetry retry = new ListenerRetry();
public boolean isAutoStartup() {
return this.autoStartup;
}
@ -429,6 +447,143 @@ public class RabbitProperties {
public void setTransactionSize(Integer transactionSize) {
this.transactionSize = transactionSize;
}
public Boolean getDefaultRequeueRejected() {
return this.defaultRequeueRejected;
}
public void setDefaultRequeueRejected(Boolean defaultRequeueRejected) {
this.defaultRequeueRejected = defaultRequeueRejected;
}
public ListenerRetry getRetry() {
return this.retry;
}
}
public static class Template {
@NestedConfigurationProperty
private final Retry retry = new Retry();
/**
* Timeout for receive() operations.
*/
private Long receiveTimeout;
/**
* Timeout for sendAndReceive() operations.
*/
private Long replyTimeout;
public Retry getRetry() {
return this.retry;
}
public Long getReceiveTimeout() {
return this.receiveTimeout;
}
public void setReceiveTimeout(Long receiveTimeout) {
this.receiveTimeout = receiveTimeout;
}
public Long getReplyTimeout() {
return this.replyTimeout;
}
public void setReplyTimeout(Long replyTimeout) {
this.replyTimeout = replyTimeout;
}
}
public static class Retry {
/**
* Whether or not publishing retries are enabled.
*/
private boolean enabled;
/**
* Maximum number of attempts to publish or deliver a message.
*/
private int maxAttempts = 3;
/**
* Interval between the first and second attempt to publish or deliver
* a message.
*/
private long initialInterval = 1000L;
/**
* A multiplier to apply to the previous retry interval.
*/
private double multiplier = 1.0;
/**
* Maximum interval between attempts.
*/
private long maxInterval = 10000L;
public boolean isEnabled() {
return this.enabled;
}
public void setEnabled(boolean enabled) {
this.enabled = enabled;
}
public int getMaxAttempts() {
return this.maxAttempts;
}
public void setMaxAttempts(int maxAttempts) {
this.maxAttempts = maxAttempts;
}
public long getInitialInterval() {
return this.initialInterval;
}
public void setInitialInterval(long initialInterval) {
this.initialInterval = initialInterval;
}
public double getMultiplier() {
return this.multiplier;
}
public void setMultiplier(double multiplier) {
this.multiplier = multiplier;
}
public long getMaxInterval() {
return this.maxInterval;
}
public void setMaxInterval(long maxInterval) {
this.maxInterval = maxInterval;
}
}
public static class ListenerRetry extends Retry {
/**
* Whether or not retries are stateless or stateful.
*/
private boolean stateless = true;
public boolean isStateless() {
return this.stateless;
}
public void setStateless(boolean stateless) {
this.stateless = stateless;
}
}
}

19
spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/SimpleRabbitListenerContainerFactoryConfigurer.java
View File

@ -16,16 +16,20 @@
package org.springframework.boot.autoconfigure.amqp;
import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties.ListenerRetry;
import org.springframework.util.Assert;
/**
* Configure {@link RabbitListenerContainerFactory} with sensible defaults.
*
* @author Stephane Nicoll
* @author Gary Russell
* @since 1.3.3
*/
public final class SimpleRabbitListenerContainerFactoryConfigurer {
@ -83,6 +87,21 @@ public final class SimpleRabbitListenerContainerFactoryConfigurer {
if (listenerConfig.getTransactionSize() != null) {
factory.setTxSize(listenerConfig.getTransactionSize());
}
if (listenerConfig.getDefaultRequeueRejected() != null) {
factory.setDefaultRequeueRejected(listenerConfig.getDefaultRequeueRejected());
}
ListenerRetry retryConfig = listenerConfig.getRetry();
if (retryConfig.isEnabled()) {
RetryInterceptorBuilder<?> builder = (retryConfig.isStateless() ?
RetryInterceptorBuilder.stateless() : RetryInterceptorBuilder.stateful());
factory.setAdviceChain(builder
.maxAttempts(retryConfig.getMaxAttempts())
.backOffOptions(retryConfig.getInitialInterval(),
retryConfig.getMultiplier(), retryConfig.getMaxInterval())
.recoverer(new RejectAndDontRequeueRecoverer())
.build());
}
}
}

65
spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/amqp/RabbitAutoConfigurationTests.java
View File

@ -19,6 +19,7 @@ package org.springframework.boot.autoconfigure.amqp;
import javax.net.SocketFactory;
import javax.net.ssl.SSLSocketFactory;
import org.aopalliance.aop.Advice;
import org.junit.After;
import org.junit.Rule;
import org.junit.Test;
@ -44,6 +45,9 @@ import org.springframework.context.annotation.AnnotationConfigApplicationContext
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
@ -88,7 +92,7 @@ public class RabbitAutoConfigurationTests {
}
@Test
public void testRabbitTemplateWithOverrides() {
public void testConnectionFactoryWithOverrides() {
load(TestConfiguration.class, "spring.rabbitmq.host:remote-server",
"spring.rabbitmq.port:9000", "spring.rabbitmq.username:alice",
"spring.rabbitmq.password:secret", "spring.rabbitmq.virtual_host:/vhost");
@ -100,7 +104,7 @@ public class RabbitAutoConfigurationTests {
}
@Test
public void testRabbitTemplateEmptyVirtualHost() {
public void testConnectionFactoryEmptyVirtualHost() {
load(TestConfiguration.class, "spring.rabbitmq.virtual_host:");
CachingConnectionFactory connectionFactory = this.context
.getBean(CachingConnectionFactory.class);
@ -108,7 +112,7 @@ public class RabbitAutoConfigurationTests {
}
@Test
public void testRabbitTemplateVirtualHostNoLeadingSlash() {
public void testConnectionFactoryVirtualHostNoLeadingSlash() {
load(TestConfiguration.class, "spring.rabbitmq.virtual_host:foo");
CachingConnectionFactory connectionFactory = this.context
.getBean(CachingConnectionFactory.class);
@ -116,7 +120,7 @@ public class RabbitAutoConfigurationTests {
}
@Test
public void testRabbitTemplateVirtualHostMultiLeadingSlashes() {
public void testConnectionFactoryVirtualHostMultiLeadingSlashes() {
load(TestConfiguration.class, "spring.rabbitmq.virtual_host:///foo");
CachingConnectionFactory connectionFactory = this.context
.getBean(CachingConnectionFactory.class);
@ -124,7 +128,7 @@ public class RabbitAutoConfigurationTests {
}
@Test
public void testRabbitTemplateDefaultVirtualHost() {
public void testConnectionFactoryDefaultVirtualHost() {
load(TestConfiguration.class, "spring.rabbitmq.virtual_host:/");
CachingConnectionFactory connectionFactory = this.context
.getBean(CachingConnectionFactory.class);
@ -137,6 +141,32 @@ public class RabbitAutoConfigurationTests {
RabbitTemplate rabbitTemplate = this.context.getBean(RabbitTemplate.class);
assertThat(rabbitTemplate.getMessageConverter())
.isSameAs(this.context.getBean("myMessageConverter"));
DirectFieldAccessor dfa = new DirectFieldAccessor(rabbitTemplate);
assertThat(dfa.getPropertyValue("retryTemplate")).isNull();
}
@Test
public void testRabbitTemplateRetry() {
load(TestConfiguration.class, "spring.rabbitmq.template.retry.enabled:true",
"spring.rabbitmq.template.retry.maxAttempts:4",
"spring.rabbitmq.template.retry.initialInterval:2000",
"spring.rabbitmq.template.retry.multiplier:1.5",
"spring.rabbitmq.template.retry.maxInterval:5000",
"spring.rabbitmq.template.receiveTimeout:123",
"spring.rabbitmq.template.replyTimeout:456");
RabbitTemplate rabbitTemplate = this.context.getBean(RabbitTemplate.class);
DirectFieldAccessor dfa = new DirectFieldAccessor(rabbitTemplate);
assertThat(dfa.getPropertyValue("receiveTimeout")).isEqualTo(123L);
assertThat(dfa.getPropertyValue("replyTimeout")).isEqualTo(456L);
RetryTemplate retryTemplate = (RetryTemplate) dfa.getPropertyValue("retryTemplate");
assertThat(retryTemplate).isNotNull();
dfa = new DirectFieldAccessor(retryTemplate);
SimpleRetryPolicy retryPolicy = (SimpleRetryPolicy) dfa.getPropertyValue("retryPolicy");
ExponentialBackOffPolicy backOffPolicy = (ExponentialBackOffPolicy) dfa.getPropertyValue("backOffPolicy");
assertThat(retryPolicy.getMaxAttempts()).isEqualTo(4);
assertThat(backOffPolicy.getInitialInterval()).isEqualTo(2000);
assertThat(backOffPolicy.getMultiplier()).isEqualTo(1.5);
assertThat(backOffPolicy.getMaxInterval()).isEqualTo(5000);
}
@Test
@ -210,16 +240,25 @@ public class RabbitAutoConfigurationTests {
SimpleRabbitListenerContainerFactory.class);
rabbitListenerContainerFactory.setTxSize(10);
verify(rabbitListenerContainerFactory).setTxSize(10);
DirectFieldAccessor dfa = new DirectFieldAccessor(rabbitListenerContainerFactory);
Advice[] adviceChain = (Advice[]) dfa.getPropertyValue("adviceChain");
assertThat(adviceChain).isNull();
}
@Test
public void testRabbitListenerContainerFactoryWithCustomSettings() {
load(MessageConvertersConfiguration.class,
"spring.rabbitmq.listener.retry.enabled:true",
"spring.rabbitmq.listener.retry.maxAttempts:4",
"spring.rabbitmq.listener.retry.initialInterval:2000",
"spring.rabbitmq.listener.retry.multiplier:1.5",
"spring.rabbitmq.listener.retry.maxInterval:5000",
"spring.rabbitmq.listener.autoStartup:false",
"spring.rabbitmq.listener.acknowledgeMode:manual",
"spring.rabbitmq.listener.concurrency:5",
"spring.rabbitmq.listener.maxConcurrency:10",
"spring.rabbitmq.listener.prefetch=40",
"spring.rabbitmq.listener.prefetch:40",
"spring.rabbitmq.listener.defaultRequeueRejected:false",
"spring.rabbitmq.listener.transactionSize:20");
SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory = this.context
.getBean("rabbitListenerContainerFactory",
@ -234,6 +273,20 @@ public class RabbitAutoConfigurationTests {
assertThat(dfa.getPropertyValue("txSize")).isEqualTo(20);
assertThat(dfa.getPropertyValue("messageConverter"))
.isSameAs(this.context.getBean("myMessageConverter"));
assertThat(dfa.getPropertyValue("defaultRequeueRejected")).isEqualTo(Boolean.FALSE);
Advice[] adviceChain = (Advice[]) dfa.getPropertyValue("adviceChain");
assertThat(adviceChain).isNotNull();
assertThat(adviceChain.length).isEqualTo(1);
dfa = new DirectFieldAccessor(adviceChain[0]);
RetryTemplate retryTemplate = (RetryTemplate) dfa.getPropertyValue("retryOperations");
assertThat(retryTemplate).isNotNull();
dfa = new DirectFieldAccessor(retryTemplate);
SimpleRetryPolicy retryPolicy = (SimpleRetryPolicy) dfa.getPropertyValue("retryPolicy");
ExponentialBackOffPolicy backOffPolicy = (ExponentialBackOffPolicy) dfa.getPropertyValue("backOffPolicy");
assertThat(retryPolicy.getMaxAttempts()).isEqualTo(4);
assertThat(backOffPolicy.getInitialInterval()).isEqualTo(2000);
assertThat(backOffPolicy.getMultiplier()).isEqualTo(1.5);
assertThat(backOffPolicy.getMaxInterval()).isEqualTo(5000);
}
@Test

14
spring-boot-docs/src/main/asciidoc/appendix-application-properties.adoc
View File

@ -768,8 +768,15 @@ content into your application; rather pick only the properties that you need.
spring.rabbitmq.listener.acknowledge-mode= # Acknowledge mode of container.
spring.rabbitmq.listener.auto-startup=true # Start the container automatically on startup.
spring.rabbitmq.listener.concurrency= # Minimum number of consumers.
spring.rabbitmq.listener.default-requeue-rejected= # Whether or not to requeue delivery failures; default `true`.
spring.rabbitmq.listener.max-concurrency= # Maximum number of consumers.
spring.rabbitmq.listener.prefetch= # Number of messages to be handled in a single request. It should be greater than or equal to the transaction size (if used).
spring.rabbitmq.listener.retry.enabled= # Whether or not publishing retries are enabled.
spring.rabbitmq.listener.retry.initial-interval=1000 # Interval between the first and second attempt to deliver a message.
spring.rabbitmq.listener.retry.max-attempts=3 # Maximum number of attempts to deliver a message.
spring.rabbitmq.listener.retry.max-interval=10000 # Maximum number of attempts to deliver a message.
spring.rabbitmq.listener.retry.multiplier=1.0 # A multiplier to apply to the previous delivery retry interval.
spring.rabbitmq.listener.retry.stateless=true # Whether or not retry is stateless or stateful.
spring.rabbitmq.listener.transaction-size= # Number of messages to be processed in a transaction. For best results it should be less than or equal to the prefetch count.
spring.rabbitmq.password= # Login to authenticate against the broker.
spring.rabbitmq.port=5672 # RabbitMQ port.
@ -779,6 +786,13 @@ content into your application; rather pick only the properties that you need.
spring.rabbitmq.ssl.key-store-password= # Password used to access the key store.
spring.rabbitmq.ssl.trust-store= # Trust store that holds SSL certificates.
spring.rabbitmq.ssl.trust-store-password= # Password used to access the trust store.
spring.rabbitmq.template.receive-timeout=0 # Timeout for `receive()` methods.
spring.rabbitmq.template.reply-timeout=5000 # Timeout for `sendAndReceive()` methods.
spring.rabbitmq.template.retry.enabled= # Set to true to enable retries in the `RabbitTemplate`.
spring.rabbitmq.template.retry.initial-interval=1000 # Interval between the first and second attempt to publish a message.
spring.rabbitmq.template.retry.max-attempts=3 # Maximum number of attempts to publish a message.
spring.rabbitmq.template.retry.max-interval=10000 # Maximum number of attempts to publish a message.
spring.rabbitmq.template.retry.multiplier=1.0 # A multiplier to apply to the previous publishing retry interval.
spring.rabbitmq.username= # Login user to authenticate to the broker.
spring.rabbitmq.virtual-host= # Virtual host to use when connecting to the broker.

13
spring-boot-docs/src/main/asciidoc/spring-boot-features.adoc
View File

@ -3801,7 +3801,8 @@ automatically to the auto-configured `AmqpTemplate`.
Any `org.springframework.amqp.core.Queue` that is defined as a bean will be automatically
used to declare a corresponding queue on the RabbitMQ instance if necessary.
You can enable retries on the `AmqpTemplate` to retry operations, for example in the event
the broker connection is lost. Retries are disabled by default.
[[boot-features-using-amqp-receiving]]
==== Receiving a message
@ -3868,6 +3869,16 @@ That you can use in any `@RabbitListener`-annotated method as follows:
}
----
You can enable retries to handle situations where your listener throws an exception.
When retries are exhausted, the message will be rejected and either dropped or routed to a dead-letter exchange
if the broker is so configured.
Retries are disabled by default.
IMPORTANT: If retries are not enabled and the listener throws an exception, by default the delivery will be retried
indefinitely.
You can modify this behavior in two ways; set the `defaultRequeueRejected` property to `false` and zero redeliveries
will be attempted; or, throw an `AmqpRejectAndDontRequeueException` to signal the message should be rejected.
This is the mechanism used when retries are enabled and the maximum delivery attempts is reached.
[[boot-features-email]]
== Sending email