From 01cb986fd11c5769f956c8da7ebfcde26b3ff0ee Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 19 Aug 2020 14:38:12 -0400 Subject: [PATCH] Add Deterministic Address Shuffling Previously, only random address shuffling was supported; it is useful, for scenarios such as using the RabbitMQ Sharding Plugin, to be able to connect to multiple nodes in a deterministic manner. --- .../config/ConnectionFactoryParser.java | 7 ++ .../connection/AbstractConnectionFactory.java | 69 +++++++++++++++---- .../amqp/rabbit/config/spring-rabbit.xsd | 24 ++++++- .../config/ConnectionFactoryParserTests.java | 3 +- .../CachingConnectionFactoryTests.java | 33 ++++++++- .../ConnectionFactoryParserTests-context.xml | 2 +- src/reference/asciidoc/amqp.adoc | 10 +-- 7 files changed, 126 insertions(+), 22 deletions(-) diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java index e7be75a3..b7e8c347 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java @@ -42,6 +42,8 @@ class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser { private static final String SHUFFLE_ADDRESSES = "shuffle-addresses"; + private static final String SHUFFLE_MODE = "address-shuffle-mode"; + private static final String ADDRESS_RESOLVER = "address-resolver"; private static final String VIRTUAL_HOST_ATTRIBUTE = "virtual-host"; @@ -103,6 +105,11 @@ class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser { NamespaceUtils.setReferenceIfAttributeDefined(builder, element, EXECUTOR_ATTRIBUTE); NamespaceUtils.setValueIfAttributeDefined(builder, element, ADDRESSES); NamespaceUtils.setValueIfAttributeDefined(builder, element, SHUFFLE_ADDRESSES); + if (element.hasAttribute(SHUFFLE_ADDRESSES) && element.hasAttribute(SHUFFLE_MODE)) { + parserContext.getReaderContext() + .error("You must not specify both '" + SHUFFLE_ADDRESSES + "' and '" + SHUFFLE_MODE + "'", element); + } + NamespaceUtils.setValueIfAttributeDefined(builder, element, SHUFFLE_MODE); NamespaceUtils.setReferenceIfAttributeDefined(builder, element, ADDRESS_RESOLVER); NamespaceUtils.setValueIfAttributeDefined(builder, element, PUBLISHER_RETURNS); NamespaceUtils.setValueIfAttributeDefined(builder, element, REQUESTED_HEARTBEAT, "requestedHeartBeat"); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java index 0e39cb99..0307a653 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java @@ -25,6 +25,7 @@ import java.security.GeneralSecurityException; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.LinkedList; import java.util.List; import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; @@ -68,6 +69,30 @@ import com.rabbitmq.client.impl.recovery.AutorecoveringConnection; public abstract class AbstractConnectionFactory implements ConnectionFactory, DisposableBean, BeanNameAware, ApplicationContextAware, ApplicationEventPublisherAware, ApplicationListener { + /** + * The mode used to shuffle the addresses. + */ + public enum AddressShuffleMode { + + /** + * Do not shuffle the addresses before or after opening a connection; attempt + * connections in a fixed order. + */ + NONE, + + /** + * Randomly shuffle the addresses before opening a connection; attempt connections + * in the new order. + */ + RANDOM, + + /** + * Shuffle the addresses after opening a connection, moving the first address to the end. + */ + INORDER + + } + private static final String PUBLISHER_SUFFIX = ".publisher"; public static final int DEFAULT_CLOSE_TIMEOUT = 30000; @@ -108,7 +133,7 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di private List
addresses; - private boolean shuffleAddresses; + private AddressShuffleMode addressShuffleMode = AddressShuffleMode.NONE; private int closeTimeout = DEFAULT_CLOSE_TIMEOUT; @@ -294,11 +319,11 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di * This property overrides the host+port properties if not empty. * @param addresses list of addresses with form "host[:port],..." */ - public void setAddresses(String addresses) { + public synchronized void setAddresses(String addresses) { if (StringUtils.hasText(addresses)) { Address[] addressArray = Address.parseAddresses(addresses); if (addressArray.length > 0) { - this.addresses = Arrays.asList(addressArray); + this.addresses = new LinkedList<>(Arrays.asList(addressArray)); if (this.publisherConnectionFactory != null) { this.publisherConnectionFactory.setAddresses(addresses); } @@ -466,9 +491,23 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di * @param shuffleAddresses true to shuffle the list. * @since 2.1.8 * @see Collections#shuffle(List) + * @deprecated since 2.3 in favor of + * {@link #setAddressShuffleMode(AddressShuffleMode)}. */ + @Deprecated public void setShuffleAddresses(boolean shuffleAddresses) { - this.shuffleAddresses = shuffleAddresses; + setAddressShuffleMode(AddressShuffleMode.RANDOM); + } + + /** + * Set the mode for shuffling addresses. + * @param addressShuffleMode the address shuffle mode. + * @since 2.3 + * @see Collections#shuffle(List) + */ + public void setAddressShuffleMode(AddressShuffleMode addressShuffleMode) { + Assert.notNull(addressShuffleMode, "'addressShuffleMode' cannot be null"); + this.addressShuffleMode = addressShuffleMode; } public boolean hasPublisherConnectionFactory() { @@ -525,7 +564,9 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di } } - private com.rabbitmq.client.Connection connect(String connectionName) throws IOException, TimeoutException { + private synchronized com.rabbitmq.client.Connection connect(String connectionName) + throws IOException, TimeoutException { + if (this.addressResolver != null) { return connectResolver(connectionName); } @@ -545,20 +586,22 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di connectionName); } - private com.rabbitmq.client.Connection connectAddresses(String connectionName) + private synchronized com.rabbitmq.client.Connection connectAddresses(String connectionName) throws IOException, TimeoutException { - List
addressesToConnect = this.addresses; - if (this.shuffleAddresses && addressesToConnect.size() > 1) { - List
list = new ArrayList<>(addressesToConnect); - Collections.shuffle(list); - addressesToConnect = list; + List
addressesToConnect = new ArrayList<>(this.addresses); + if (addressesToConnect.size() > 1 && AddressShuffleMode.RANDOM.equals(this.addressShuffleMode)) { + Collections.shuffle(addressesToConnect); } if (this.logger.isInfoEnabled()) { this.logger.info("Attempting to connect to: " + addressesToConnect); } - return this.rabbitConnectionFactory.newConnection(this.executorService, addressesToConnect, - connectionName); + com.rabbitmq.client.Connection connection = this.rabbitConnectionFactory.newConnection(this.executorService, + addressesToConnect, connectionName); + if (addressesToConnect.size() > 1 && AddressShuffleMode.INORDER.equals(this.addressShuffleMode)) { + this.addresses.add(this.addresses.remove(0)); + } + return connection; } private com.rabbitmq.client.Connection connectHostPort(String connectionName) throws IOException, TimeoutException { diff --git a/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit.xsd b/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit.xsd index da6eb6b7..bb308496 100644 --- a/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit.xsd +++ b/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit.xsd @@ -1417,7 +1417,7 @@ @@ -1425,9 +1425,23 @@ + + + + + + + + + + + + + + + + diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests.java index 073a2d67..20256ff2 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests.java @@ -24,6 +24,7 @@ import java.util.concurrent.ExecutorService; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory.AddressShuffleMode; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType; import org.springframework.amqp.rabbit.connection.ConnectionNameStrategy; @@ -123,7 +124,7 @@ public final class ConnectionFactoryParserTests { assertThat(addresses.get(1).getPort()).isEqualTo(-1); assertThat(addresses.get(2).getHost()).isEqualTo("host3"); assertThat(addresses.get(2).getPort()).isEqualTo(4567); - assertThat(dfa.getPropertyValue("shuffleAddresses")).isEqualTo(Boolean.TRUE); + assertThat(dfa.getPropertyValue("addressShuffleMode")).isEqualTo(AddressShuffleMode.INORDER); assertThat(TestUtils.getPropertyValue(connectionFactory, "rabbitConnectionFactory.threadFactory")).isSameAs(beanFactory.getBean("tf")); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java index 38b4cdfe..f15a404a 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java @@ -70,6 +70,7 @@ import org.mockito.InOrder; import org.springframework.amqp.AmqpConnectException; import org.springframework.amqp.AmqpTimeoutException; +import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory.AddressShuffleMode; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.CacheMode; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType; import org.springframework.amqp.rabbit.core.RabbitTemplate; @@ -1817,7 +1818,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest @SuppressWarnings("unchecked") @Test - public void testShuffle() throws IOException, TimeoutException { + public void testShuffleRandom() throws IOException, TimeoutException { com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class); com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class); Channel mockChannel = mock(Channel.class); @@ -1831,7 +1832,7 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); ccf.setCacheMode(CacheMode.CONNECTION); ccf.setAddresses("host1:5672,host2:5672,host3:5672"); - ccf.setShuffleAddresses(true); + ccf.setAddressShuffleMode(AddressShuffleMode.RANDOM); IntStream.range(0, 100).forEach(i -> ccf.createConnection()); ccf.destroy(); ArgumentCaptor> captor = ArgumentCaptor.forClass(List.class); @@ -1845,6 +1846,34 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest assertThat(firstAddress).containsExactly("host1", "host2", "host3"); } + @SuppressWarnings("unchecked") + @Test + public void testShuffleInOrder() throws IOException, TimeoutException { + com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class); + com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class); + Channel mockChannel = mock(Channel.class); + + given(mockConnectionFactory.newConnection((ExecutorService) isNull(), any(List.class), anyString())) + .willReturn(mockConnection); + given(mockConnection.createChannel()).willReturn(mockChannel); + given(mockChannel.isOpen()).willReturn(true); + given(mockConnection.isOpen()).willReturn(true); + + CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); + ccf.setCacheMode(CacheMode.CONNECTION); + ccf.setAddresses("host1:5672,host2:5672,host3:5672"); + ccf.setAddressShuffleMode(AddressShuffleMode.INORDER); + IntStream.range(0, 3).forEach(i -> ccf.createConnection()); + ccf.destroy(); + ArgumentCaptor> captor = ArgumentCaptor.forClass(List.class); + verify(mockConnectionFactory, times(3)).newConnection(isNull(), captor.capture(), anyString()); + List connectAddresses = captor.getAllValues() + .stream() + .map(addresses -> addresses.get(0).getHost()) + .collect(Collectors.toList()); + assertThat(connectAddresses).containsExactly("host1", "host2", "host3"); + } + @Test void testResolver() throws Exception { com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class); diff --git a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml index 6e2d7939..4ae8d756 100644 --- a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml +++ b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml @@ -45,7 +45,7 @@ diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 3bda9d2f..40a7968c 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -412,11 +412,11 @@ Alternatively, if running in a clustered environment, you can use the addresses [source,xml] ---- + id="connectionFactory" addresses="host1:5672,host2:5672" address-shuffle-mode="RANDOM"/> ---- ==== -See <> for information about `shuffle-addresses`. +See <> for information about `address-shuffle-mode`. The following example with a custom thread factory that prefixes thread names with `rabbitmq-`: @@ -605,7 +605,9 @@ public CachingConnectionFactory ccf() { ==== The underlying connection factory will attempt to connect to each host, in order, whenever a new connection is established. -Starting with version 2.1.8, the connection order can be made random by setting the `shuffleAddresses` property to true; the shuffle will be applied before creating any new connection. +Starting with version 2.1.8, the connection order can be made random by setting the `addressShuffleMode` property to `RANDOM`; the shuffle will be applied before creating any new connection. +Starting with version 2.6, the `INORDER` shuffle mode was added, which means the first address is moved to the end after a connection is created. +You may wish to use this mode with the https://github.com/rabbitmq/rabbitmq-sharding[RabbitMQ Sharding Plugin] with `CacheMode.CONNECTION` and suitable concurrency if you wish to consume from all shards on all nodes. ==== [source, java] @@ -614,7 +616,7 @@ Starting with version 2.1.8, the connection order can be made random by setting public CachingConnectionFactory ccf() { CachingConnectionFactory ccf = new CachingConnectionFactory(); ccf.setAddresses("host1:5672,host2:5672,host3:5672"); - ccf.setShuffleAddresses(true); + ccf.setAddressShuffleMode(AddressShuffleMode.RANDOM); return ccf; } ----