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 206cf9afa..2dfa2cb66 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 @@ -44,7 +44,7 @@ If retry is enabled (`maxAttempts > 1`) failed messages will be delivered to the 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 messsages; see property `republishDeliveryMode`. +Starting with _version 1.2_, you can configure the delivery mode of republished messages; see property `republishDeliveryMode`. IMPORTANT: Setting `requeueRejected` to `true` 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's better to enable retry within the binder by setting `maxAttempts` to greater than one, or set `republishToDlq` to `true`. @@ -56,9 +56,11 @@ 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. +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. ==== +Starting with _version 1.3_, the `RabbitMessageChannelBinder` creates an internal `ConnectionFactory` copy for the non-transactional producers to avoid dead locks on consumers when shared, cached connections are blocked because of https://www.rabbitmq.com/memory.html[Memory Alarm] on Broker. + == Configuration Options This section contains settings specific to the RabbitMQ Binder and bound channels. diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index 4cdcd1086..6fab833c2 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -30,6 +30,7 @@ import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.connection.LocalizedQueueConnectionFactory; +import org.springframework.amqp.rabbit.connection.RabbitConnectionFactoryBean; import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.core.support.BatchingStrategy; @@ -42,7 +43,7 @@ import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter import org.springframework.amqp.rabbit.support.MessagePropertiesConverter; import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor; import org.springframework.amqp.support.postprocessor.GZipPostProcessor; -import org.springframework.beans.factory.NoSuchBeanDefinitionException; +import org.springframework.beans.factory.DisposableBean; import org.springframework.boot.autoconfigure.amqp.RabbitProperties; import org.springframework.boot.autoconfigure.amqp.RabbitProperties.Retry; import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder; @@ -82,7 +83,9 @@ import com.rabbitmq.client.AMQP; import com.rabbitmq.client.Envelope; /** - * A {@link org.springframework.cloud.stream.binder.Binder} implementation backed by RabbitMQ. + * A {@link org.springframework.cloud.stream.binder.Binder} implementation backed by + * RabbitMQ. + * * @author Mark Fisher * @author Gary Russell * @author Jennifer Hickey @@ -90,11 +93,13 @@ import com.rabbitmq.client.Envelope; * @author Ilayaperumal Gopinathan * @author David Turanski * @author Marius Bogoevici + * @author Artem Bilan */ public class RabbitMessageChannelBinder extends AbstractMessageChannelBinder, ExtendedProducerProperties, RabbitExchangeQueueProvisioner> - implements ExtendedPropertiesBinder { + implements ExtendedPropertiesBinder, + DisposableBean { private static final AmqpMessageHeaderErrorMessageStrategy errorMessageStrategy = new AmqpMessageHeaderErrorMessageStrategy(); @@ -109,12 +114,15 @@ public class RabbitMessageChannelBinder properties.setDeliveryMode(null); return properties; } + }; private final RabbitProperties rabbitProperties; private ConnectionFactory connectionFactory; + private ConnectionFactory producerConnectionFactory; + private MessagePostProcessor decompressingPostProcessor = new DelegatingDecompressingPostProcessor(); private MessagePostProcessor compressingPostProcessor = new GZipPostProcessor(); @@ -128,7 +136,7 @@ public class RabbitMessageChannelBinder private RabbitExtendedBindingProperties extendedBindingProperties = new RabbitExtendedBindingProperties(); public RabbitMessageChannelBinder(ConnectionFactory connectionFactory, RabbitProperties rabbitProperties, - RabbitExchangeQueueProvisioner provisioningProvider) { + RabbitExchangeQueueProvisioner provisioningProvider) { super(true, new String[0], provisioningProvider); Assert.notNull(connectionFactory, "connectionFactory must not be null"); Assert.notNull(rabbitProperties, "rabbitProperties must not be null"); @@ -146,8 +154,8 @@ public class RabbitMessageChannelBinder } /** - * Set a {@link org.springframework.amqp.core.MessagePostProcessor} to compress messages. Defaults to a - * {@link org.springframework.amqp.support.postprocessor.GZipPostProcessor}. + * Set a {@link org.springframework.amqp.core.MessagePostProcessor} to compress messages. + * Defaults to a {@link org.springframework.amqp.support.postprocessor.GZipPostProcessor}. * @param compressingPostProcessor the post processor. */ public void setCompressingPostProcessor(MessagePostProcessor compressingPostProcessor) { @@ -168,12 +176,23 @@ public class RabbitMessageChannelBinder } @Override - public void onInit() { + public void onInit() throws Exception { + super.onInit(); + + CachingConnectionFactory producerConnectionFactory = createProducerConnectionFactory(this.rabbitProperties); + producerConnectionFactory.setApplicationContext(getApplicationContext()); + getApplicationContext().addApplicationListener(producerConnectionFactory); + producerConnectionFactory.afterPropertiesSet(); + + this.producerConnectionFactory = producerConnectionFactory; + if (this.clustered) { String[] addresses = StringUtils.commaDelimitedListToStringArray(this.rabbitProperties.getAddresses()); + Assert.state(addresses.length == this.adminAddresses.length - && addresses.length == this.nodes.length, + && addresses.length == this.nodes.length, "'addresses', 'adminAddresses', and 'nodes' properties must have equal length"); + this.connectionFactory = new LocalizedQueueConnectionFactory(this.connectionFactory, addresses, this.adminAddresses, this.nodes, this.rabbitProperties.getVirtualHost(), this.rabbitProperties.getUsername(), this.rabbitProperties.getPassword(), @@ -184,6 +203,82 @@ public class RabbitMessageChannelBinder } } + /** + * @see org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration.RabbitConnectionFactoryCreator + */ + private CachingConnectionFactory createProducerConnectionFactory(RabbitProperties config) throws Exception { + com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory = null; + if (this.connectionFactory instanceof CachingConnectionFactory) { + rabbitConnectionFactory = ((CachingConnectionFactory) this.connectionFactory).getRabbitConnectionFactory(); + } + else { + RabbitConnectionFactoryBean factory = new RabbitConnectionFactoryBean(); + if (config.determineHost() != null) { + factory.setHost(config.determineHost()); + } + factory.setPort(config.determinePort()); + if (config.determineUsername() != null) { + factory.setUsername(config.determineUsername()); + } + if (config.determinePassword() != null) { + factory.setPassword(config.determinePassword()); + } + if (config.determineVirtualHost() != null) { + factory.setVirtualHost(config.determineVirtualHost()); + } + if (config.getRequestedHeartbeat() != null) { + factory.setRequestedHeartbeat(config.getRequestedHeartbeat()); + } + RabbitProperties.Ssl ssl = config.getSsl(); + if (ssl.isEnabled()) { + factory.setUseSSL(true); + if (ssl.getAlgorithm() != null) { + factory.setSslAlgorithm(ssl.getAlgorithm()); + } + factory.setKeyStore(ssl.getKeyStore()); + factory.setKeyStorePassphrase(ssl.getKeyStorePassword()); + factory.setTrustStore(ssl.getTrustStore()); + factory.setTrustStorePassphrase(ssl.getTrustStorePassword()); + } + if (config.getConnectionTimeout() != null) { + factory.setConnectionTimeout(config.getConnectionTimeout()); + } + factory.afterPropertiesSet(); + + rabbitConnectionFactory = factory.getObject(); + } + + CachingConnectionFactory connectionFactory = new CachingConnectionFactory(rabbitConnectionFactory); + connectionFactory.setAddresses(config.determineAddresses()); + connectionFactory.setPublisherConfirms(config.isPublisherConfirms()); + connectionFactory.setPublisherReturns(config.isPublisherReturns()); + if (config.getCache().getChannel().getSize() != null) { + connectionFactory + .setChannelCacheSize(config.getCache().getChannel().getSize()); + } + if (config.getCache().getConnection().getMode() != null) { + connectionFactory + .setCacheMode(config.getCache().getConnection().getMode()); + } + if (config.getCache().getConnection().getSize() != null) { + connectionFactory.setConnectionCacheSize( + config.getCache().getConnection().getSize()); + } + if (config.getCache().getChannel().getCheckoutTimeout() != null) { + connectionFactory.setChannelCheckoutTimeout( + config.getCache().getChannel().getCheckoutTimeout()); + } + return connectionFactory; + } + + @Override + public void destroy() throws Exception { + if (this.connectionFactory instanceof DisposableBean) { + ((DisposableBean) this.connectionFactory).destroy(); + ((DisposableBean) this.producerConnectionFactory).destroy(); + } + } + @Override public RabbitConsumerProperties getExtendedConsumerProperties(String channelName) { return this.extendedBindingProperties.getExtendedConsumerProperties(channelName); @@ -197,7 +292,7 @@ public class RabbitMessageChannelBinder @Override protected MessageHandler createProducerMessageHandler(final ProducerDestination producerDestination, ExtendedProducerProperties producerProperties, MessageChannel errorChannel) - throws Exception { + throws Exception { String prefix = producerProperties.getExtension().getPrefix(); String exchangeName = producerDestination.getName(); String destination = StringUtils.isEmpty(prefix) ? exchangeName : exchangeName.substring(prefix.length()); @@ -274,7 +369,7 @@ public class RabbitMessageChannelBinder @Override protected MessageProducer createConsumerEndpoint(ConsumerDestination consumerDestination, String group, - ExtendedConsumerProperties properties) { + ExtendedConsumerProperties properties) { String destination = consumerDestination.getName(); SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer( this.connectionFactory); @@ -330,7 +425,7 @@ public class RabbitMessageChannelBinder return new MessageHandler() { private final RabbitTemplate template = new RabbitTemplate( - RabbitMessageChannelBinder.this.connectionFactory); + RabbitMessageChannelBinder.this.producerConnectionFactory); private final String exchange = deadLetterExchangeName(properties.getExtension()); @@ -423,13 +518,6 @@ public class RabbitMessageChannelBinder } private RabbitTemplate buildRabbitTemplate(RabbitProducerProperties properties, boolean mandatory) { - RabbitProperties rabbitProperties = null; - try { - rabbitProperties = getApplicationContext().getBean(RabbitProperties.class); - } - catch (NoSuchBeanDefinitionException e) { - logger.debug("No RabbitProperties in context; no producer retry will be configured"); - } RabbitTemplate rabbitTemplate; if (properties.isBatchingEnabled()) { BatchingStrategy batchingStrategy = new SimpleBatchingStrategy( @@ -443,11 +531,16 @@ public class RabbitMessageChannelBinder else { rabbitTemplate = new RabbitTemplate(); } - rabbitTemplate.setConnectionFactory(this.connectionFactory); + rabbitTemplate.setChannelTransacted(properties.isTransacted()); + if (rabbitTemplate.isChannelTransacted()) { + rabbitTemplate.setConnectionFactory(this.connectionFactory); + } + else { + rabbitTemplate.setConnectionFactory(this.producerConnectionFactory); + } if (properties.isCompress()) { rabbitTemplate.setBeforePublishPostProcessors(this.compressingPostProcessor); } - rabbitTemplate.setChannelTransacted(properties.isTransacted()); rabbitTemplate.setMandatory(mandatory); // returned messages if (rabbitProperties != null && rabbitProperties.getTemplate().getRetry().isEnabled()) { Retry retry = rabbitProperties.getTemplate().getRetry(); diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index ef5c89d78..31f3cb7bf 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java @@ -43,6 +43,7 @@ import org.springframework.amqp.core.ExchangeTypes; import org.springframework.amqp.core.MessageDeliveryMode; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitManagementTemplate; import org.springframework.amqp.rabbit.core.RabbitTemplate; @@ -120,10 +121,13 @@ public class RabbitBinderTests extends @Override protected RabbitTestBinder getBinder() { - if (testBinder == null) { - testBinder = new RabbitTestBinder(rabbitAvailableRule.getResource(), new RabbitProperties()); + if (this.testBinder == null) { + RabbitProperties rabbitProperties = new RabbitProperties(); + rabbitProperties.setPublisherConfirms(true); + rabbitProperties.setPublisherReturns(true); + this.testBinder = new RabbitTestBinder(rabbitAvailableRule.getResource(), rabbitProperties); } - return testBinder; + return this.testBinder; } @Override @@ -155,6 +159,20 @@ public class RabbitBinderTests extends createProducerProperties()); Binding consumerBinding = binder.bindConsumer("bad.0", "test", moduleInputChannel, createConsumerProperties()); + + ConnectionFactory producerConnectionFactory = + TestUtils.getPropertyValue(producerBinding, "lifecycle.amqpTemplate.connectionFactory", + ConnectionFactory.class); + + ConnectionFactory consumerConnectionFactory = + TestUtils.getPropertyValue(consumerBinding, "lifecycle.messageListenerContainer.connectionFactory", + ConnectionFactory.class); + + assertThat(producerConnectionFactory).isNotSameAs(consumerConnectionFactory); + + assertThat(producerConnectionFactory.createConnection()) + .isNotEqualTo(consumerConnectionFactory.createConnection()); + Message message = MessageBuilder.withPayload("bad").setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar") .build(); final CountDownLatch latch = new CountDownLatch(3); @@ -496,8 +514,15 @@ public class RabbitBinderTests extends BindingProperties producerBindingProperties = createProducerBindingProperties(producerProperties); DirectChannel channel = createBindableChannel("output", producerBindingProperties); - producerBinding = binder.bindProducer("props.0", channel, - producerProperties); + producerBinding = binder.bindProducer("props.0", channel, producerProperties); + + ConnectionFactory producerConnectionFactory = + TestUtils.getPropertyValue(producerBinding, "lifecycle.amqpTemplate.connectionFactory", + ConnectionFactory.class); + + assertThat(this.rabbitAvailableRule.getResource()) + .isSameAs(producerConnectionFactory); + endpoint = extractEndpoint(producerBinding); assertThat(getEndpointRouting(endpoint)) .isEqualTo("'props.0-' + headers['" + BinderHeaders.PARTITION_HEADER + "']"); @@ -726,12 +751,12 @@ public class RabbitBinderTests extends } @Test - public void testAutoBindDLQPartionedConsumerFirstWithRepublishNoRetry() throws Exception { + public void testAutoBindDLQPartitionedConsumerFirstWithRepublishNoRetry() throws Exception { testAutoBindDLQPartionedConsumerFirstWithRepublishGuts(false); } @Test - public void testAutoBindDLQPartionedConsumerFirstWithRepublishWithRetry() throws Exception { + public void testAutoBindDLQPartitionedConsumerFirstWithRepublishWithRetry() throws Exception { testAutoBindDLQPartionedConsumerFirstWithRepublishGuts(true); } @@ -1076,6 +1101,7 @@ public class RabbitBinderTests extends ExtendedProducerProperties producerProperties = createProducerProperties(); producerProperties.getExtension().setPrefix("latebinder."); producerProperties.getExtension().setAutoBindDlq(true); + producerProperties.getExtension().setTransacted(true); MessageChannel moduleOutputChannel = createBindableChannel("output", createProducerBindingProperties(producerProperties)); Binding late0ProducerBinding = binder.bindProducer("late.0", moduleOutputChannel, producerProperties); @@ -1135,7 +1161,7 @@ public class RabbitBinderTests extends proxy.start(); moduleOutputChannel.send(new GenericMessage<>("foo")); - Message message = moduleInputChannel.receive(10000); + Message message = moduleInputChannel.receive(20000); assertThat(message).isNotNull(); assertThat(message.getPayload()).isNotNull();