Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add Deterministic Address Shuffling

Browse Source 
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.
This commit is contained in:
Gary Russell
2020-08-19 14:38:12 -04:00
committed by Artem Bilan
parent 90e3232045
commit 01cb986fd1
7 changed files with 126 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java
View File

@@ -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");

69
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java
View File

@@ -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<ContextClosedEvent> {
/**
* 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<Address> 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<Address> addressesToConnect = this.addresses;
if (this.shuffleAddresses && addressesToConnect.size() > 1) {
List<Address> list = new ArrayList<>(addressesToConnect);
Collections.shuffle(list);
addressesToConnect = list;
List<Address> 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 {

24
spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit.xsd
View File

@@ -1417,7 +1417,7 @@
<xsd:annotation>
<xsd:documentation><![CDATA[
List of addresses; e.g. host1,host2:4567,host3 - overrides host/port if supplied.
Connection will be attempted in order unless 'shuffle-addresses' is 'true'.
Connection will be attempted in order unless 'address-shuffle-mode' is set to other than 'NONE'.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -1425,9 +1425,23 @@
<xsd:annotation>
<xsd:documentation><![CDATA[
Set to true when 'addresses' has more than one address to shuffle the list into a random order.
DEPRECATED: use 'address-shuffle-mode' instead.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="address-shuffle-mode" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
When multiple addresses are provided, determines whether or not the addresses are shuffled before
or after creating a connection. NONE means no shuffling is performed. RANDOM means that the
addresses are randomly shuffled before each connection is created. INORDER means that the first
address is moved to the end after each connection is created.
]]></xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="shuffleModes xsd:string"/>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="address-resolver" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
@@ -1642,6 +1656,14 @@
</xsd:restriction>
</xsd:simpleType>
<xsd:simpleType name="shuffleModes">
<xsd:restriction base="xsd:token">
<xsd:enumeration value="NONE"/>
<xsd:enumeration value="RANDOM"/>
<xsd:enumeration value="INORDER"/>
</xsd:restriction>
</xsd:simpleType>
<xsd:simpleType name="confirmTypes">
<xsd:restriction base="xsd:token">
<xsd:enumeration value="SIMPLE"/>

3
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests.java
View File

@@ -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"));
}

33
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java
View File

@@ -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<List<Address>> 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<List<Address>> captor = ArgumentCaptor.forClass(List.class);
verify(mockConnectionFactory, times(3)).newConnection(isNull(), captor.capture(), anyString());
List<String> 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);

2
spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml
View File

@@ -45,7 +45,7 @@
<bean id="execService" class="java.util.concurrent.Executors" factory-method="newSingleThreadExecutor" />
<rabbit:connection-factory id="multiHost" virtual-host="/bar" addresses="host1:1234,host2,host3:4567"
thread-factory="tf" shuffle-addresses="true"
thread-factory="tf" address-shuffle-mode="INORDER"
channel-cache-size="10" username="user" password="password" />
<bean id="tf" class="org.springframework.scheduling.concurrent.CustomizableThreadFactory">

10
src/reference/asciidoc/amqp.adoc
View File

@@ -412,11 +412,11 @@ Alternatively, if running in a clustered environment, you can use the addresses
[source,xml]
----
<rabbit:connection-factory
id="connectionFactory" addresses="host1:5672,host2:5672" shuffle-addresses="true"/>
id="connectionFactory" addresses="host1:5672,host2:5672" address-shuffle-mode="RANDOM"/>
----
====
See <<cluster>> for information about `shuffle-addresses`.
See <<cluster>> 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;
}
----