Chapter 20. Clustering

Sometimes using UDP is not an option due to constraints put on the network. In these situations, you can configure the broadcast group to communicate using JGroups as an alternative to UDP.

<configuration>
  <core>
    ...
    <broadcast-groups>
       <broadcast-group name="my-broadcast-group"> 1
          <jgroups-file>test-jgroups-file_ping.xml</jgroups-file> 2
          <jgroups-channel>activemq_broadcast_channel</jgroups-channel> 3
          <broadcast-period>2000</broadcast-period> 4
        <connector-ref>netty-connector</connector-ref> 5
       </broadcast-group>
    </broadcast-groups>
  </core>
</configuration>

To use JGroups to broadcast, you must specify the following:

Define discovery groups in the BROKER_INSTANCE_DIR/etc/broker.xml configuration file. You can define multiple discovery groups.

<configuration>
  <core>
    ...
    <discovery-groups>
       <discovery-group name="my-discovery-group"> 1
          <local-bind-address>172.16.9.7</local-bind-address> 2
          <group-address>231.7.7.7</group-address> 3
          <group-port>9876</group-port> 4
          <refresh-timeout>10000</refresh-timeout> 5
       </discovery-group>
    </discovery-groups>
    ...
  </core>
</configuration>

Sometimes using UDP is not an option due to constraints put on the network. In these situations, you can configure the broadcast group to communicate using JGroups as an alternative to UDP. Define discovery groups in the BROKER_INSTANCE_DIR/etc/broker.xml configuration file. You can define multiple discovery groups.

<configuration>
  <core>
    ...
    <discovery-groups>
       <discovery-group name="my-discovery-group">1
          <jgroups-file>test-jgroups-file_ping.xml</jgroups-file>2
          <jgroups-channel>activemq_broadcast_channel</jgroups-channel>3
          <refresh-timeout>10000</refresh-timeout>4
       </discovery-group>
    </discovery-groups>
    ...
  </core>
</configuration>

Configure a cluster connections by adding a cluster-connection element to BROKER_INSTANCE_DIR/etc/broker.xml.

<configuration>
  <core>
    ...
    <cluster-connections> 1
       <cluster-connection name="my-cluster"> 2
         <connector-ref>netty-connector</connector-ref> 3
         <discovery-group-ref discovery-group-name="my-discovery-group"/> 4
         ...
       </cluster-connection>
       ...
    </cluster-connections>
  </core>
</configuration>

Rather than use dynamic discovery to form a cluster, you can configure a static list of connectors that restricts the cluster to a limited set of brokers. Use this static discovery method to form non-symmetrical clusters such as chain clusters or ring clusters.

You can configure a Red Hat AMQ Core JMS client to discover a list of brokers when attempting to establish a connection.

Configuring Dynamic Discovery Using JMS

If you are using JNDI on the client to look up your JMS connection factory instances, you can specify these parameters in the JNDI context environment. Typically the parameters are defined in a file named jndi.properties. The host and part in the URL for the connection factory should match the group-address and group-port from the corresponding broadcast-group inside broker’s broker.xml configuration file. Below is an example of a jndi.properties file configured to connect to a broker’s discovery group.

java.naming.factory.initial = ActiveMQInitialContextFactory
connectionFactory.myConnectionFactory=udp://231.7.7.7:9876

When this connection factory is downloaded from JNDI by a client application and JMS connections are created from it, those connections will be load-balanced across the list of servers that the discovery group maintains by listening on the multicast address specified in the broker’s discovery group configuration.

As an alternative to using JNDI, you can use specify the discovery group parameters directly in your Java code when creating the JMS connection factory. The code below provides an example of how to do this.

final String groupAddress = "231.7.7.7";
final int groupPort = 9876;

DiscoveryGroupConfiguration discoveryGroupConfiguration = new DiscoveryGroupConfiguration();
UDPBroadcastEndpointFactory udpBroadcastEndpointFactory = new UDPBroadcastEndpointFactory();
udpBroadcastEndpointFactory.setGroupAddress(groupAddress).setGroupPort(groupPort);
discoveryGroupConfiguration.setBroadcastEndpointFactory(udpBroadcastEndpointFactory);

ConnectionFactory jmsConnectionFactory = ActiveMQJMSClient.createConnectionFactoryWithHA(discoveryGroupConfiguration, JMSFactoryType.CF);

Connection jmsConnection1 = jmsConnectionFactory.createConnection();
Connection jmsConnection2 = jmsConnectionFactory.createConnection();

The refresh timeout can be set directly on the DiscoveryGroupConfiguration by using the setter method setRefreshTimeout(). The default value is 10000 milliseconds.

On first usage, the connection factory will make sure it waits this long since creation before creating the first connection. The default wait time is 10000 milliseconds, but you can change it by passing a new value to DiscoveryGroupConfiguration.setDiscoveryInitialWaitTimeout().

Sometimes it may be impossible to use UDP on the network you are using. In this case you can configure a connection with an initial list if possible servers. The list can be just one broker that you know will always be available, or a list of brokers where at least one will be available.

This does not mean that you have to know where all your servers are going to be hosted, you can configure these servers to use the reliable servers to connect to. After they are connected, their connection details will be propagated via the server the client.

Both Red Hat AMQ Core JMS and Java EE JMS clients can use a static list to discover brokers.

Configuring Static Discovery

If you are using JNDI on the client to look up your JMS connection factory instances, you can specify these parameters in the JNDI context environment. Typically the parameters are defined in a file named jndi.properties. Below is an example jndi.properties file that provides a static list of brokers instead of using dynamic discovery.

java.naming.factory.initial=org.apache.activemq.artemis.jndi.ActiveMQInitialContextFactory
connectionFactory.myConnectionFactory=(tcp://myhost:61616,tcp://myhost2:61616)

When the above connection factory is used by a client, its connections will be load-balanced across the list of brokers defined within the parentheses ().

If you are instantiating the JMS connection factory directly, you can specify the connector list explicitly when creating the JMS connection factory, as in the example below.

HashMap<String, Object> map = new HashMap<String, Object>();
map.put("host", "myhost");
map.put("port", "61616");
TransportConfiguration broker1 = new TransportConfiguration(NettyConnectorFactory.class.getName(), map);

HashMap<String, Object> map2 = new HashMap<String, Object>();
map2.put("host", "myhost2");
map2.put("port", "61617");
TransportConfiguration broker2 = new TransportConfiguration(NettyConnectorFactory.class.getName(), map2);

ActiveMQConnectionFactory cf = ActiveMQJMSClient.createConnectionFactoryWithHA(JMSFactoryType.CF, broker1, broker2);

You can use a JNDI context environment to configure a connection factory with a load balancing policy. Using JNDI is the recommended method because the context environment can be maintained in a text file that is external to your source code. Consequently, you do not need to recompile your client application after making changes to your load balancing policy.

Procedure

You can use the AMQ Core Protocol JMS client API to programmatically set the client-side load balancing policy used by a connection factory. A disadvantage to using the API is that you must recompile your client code before any configuration changes can take effect. It is recommended that you use JNDI to configure client-side load balancing instead. For more information, see Section 20.6.1, “Setting the Load Balancing Policy By Using JNDI”.

Procedure

With a symmetric cluster, every broker in the cluster is connected to every other broker in the cluster. This means that every broker in the cluster is no more than one hop away from every other broker.

To form a symmetric cluster, every broker in the cluster defines a cluster connection with the attribute max-hops set to 1. Typically, the cluster connection will use broker discovery in order to know what other brokers in the cluster it should connect to, although it is possible to explicitly define each target broker too in the cluster connection if, for example, UDP is not available on your network.

With a symmetric cluster, each broker knows about all the queues that exist on all of the other brokers, and what consumers they have. With this knowledge, it can determine how to load balance and redistribute messages around the brokers.

With a chain cluster, each broker in the cluster is not connected to every broker in the cluster directly. Instead, the brokers form a chain with a broker on each end of the chain and all other brokers just connecting to the previous and next brokers in the chain.

An example of a chain cluster would be a three broker chain consisting of brokers A, B and C. broker A is hosted in one network and has many producer clients connected to it sending order messages. Due to corporate policy, the order consumer clients need to be hosted in a different network, and that network is only accessible via a third network. In this setup, broker B acts as a mediator with no producers or consumers on it. Any messages arriving on broker A will be forwarded to broker B, which in turn forwards them to broker C where they can be consumed. broker A does not need to directly connect to C, but all of the brokers can still act as a part of the cluster.

To set up a cluster in this way, broker A would define a cluster connection that connects to broker B, and broker B would define a cluster connection that connects to broker C. In this case, the cluster connections only need to be in one direction, because messages are moving from broker A→B→C and never from C→B→A.

For this topology, you would set max-hops to 2. With a value of 2, the knowledge of what queues and consumers that exist on broker C would be propagated from broker C to broker B to broker A. broker A would then know to distribute messages to broker B when they arrive, even though broker B has no consumers itself. It would know that a further hop away is broker C, which does have consumers.

If the size of a cluster changes frequently in your environment, you can scale it up or down with no message loss (even for non-durable messages).

You can scale up clusters by adding brokers, in which case there is no risk of message loss. Similarly, you can scale clusters down by removing brokers. However, to prevent message loss, you must first configure the broker to send its messages to another broker in the cluster.

To scale down a cluster: