Chapter 8. Flow control

Flow control prevents producers and consumers from becoming overburdened by limiting the flow of data between them. AMQ Core Protocol JMS allows you to configure flow control for both consumers and producers.

Consumer flow control

Consumer flow control regulates the flow of data between the broker and the client as the client consumes messages from the broker. AMQ Core Protocol JMS buffers messages by default before delivering them to consumers. Without a buffer, the client would first need to request each message from the broker before consuming it. This type of "round-trip" communication is costly. Regulating the flow of data on the client side is important because out of memory issues can result when a consumer cannot process messages quickly enough and the buffer begins to overflow with incoming messages.

Producer flow control

In a similar way to consumer window-based flow control, the client can limit the amount of data sent from a producer to a broker to prevent the broker from being overburdened with too much data. In the case of a producer, the window size determines the number of bytes that can be in flight at any one time.

8.1. Setting the consumer window size

The maximum size of messages held in the client-side buffer is determined by its window size. The default size of the window for AMQ Core Protocol JMS is 1 MiB, or 1024 * 1024 bytes. The default is fine for most use cases. For other cases, finding the optimal value for the window size might require benchmarking your system. AMQ Core Protocol JMS allows you to set the buffer window size if you need to change the default.

The following examples show how to set the consumer window size parameter when using AMQ Core Protocol JMS. Each example sets the consumer window size to 300,000 bytes.

Procedure

  • If the client uses JNDI to instantiate its connection factory, include the consumerWindowSize parameter as part of the connection string URL. Store the URL within a JNDI context environment. The example below uses a jndi.properties file to store the URL. 

    java.naming.factory.initial=org.apache.activemq.artemis.jndi.ActiveMQInitialContextFactory
connectionFactory.myConnectionFactory=tcp://localhost:61616?consumerWindowSize=300000

  • If the client does not use JNDI to instantiate its connection factory, pass a value to ActiveMQConnectionFactory.setConsumerWindowSize(). 

    ConnectionFactory cf =  ActiveMQJMSClient.createConnectionFactory(...)
cf.setConsumerWindowSize(300000);

8.2. Setting the producer window size

The window size is negotiated between the broker and producer on the basis of credits, one credit for each byte in the window. As messages are sent and credits are used, the producer must request, and be granted, credits from the broker before it can send more messages. The exchange of credits between producer and broker regulates the flow of data between them.

The following examples show how to set the producer window size to 1024 bytes when using AMQ Core Protocol JMS.

Procedure

  • If the client uses JNDI to instantiate its connection factory, include the producerWindowSize parameter as part of the connection string URL. Store the URL within a JNDI context environment. The example below uses a jndi.properties file to store the URL. 

    java.naming.factory.initial=org.apache.activemq.artemis.jndi.ActiveMQInitialContextFactory
java.naming.provider.url=tcp://localhost:61616?producerWindowSize=1024

  • If the client does not use JNDI to instantiate its connection factory, pass the value to ActiveMQConnectionFactory.setProducerWindowSize(). 

    ConnectionFactory cf =  ActiveMQJMSClient.createConnectionFactory(...)
cf.setProducerWindowSize(1024);

8.3. Handling fast consumers

Fast consumers can process messages as fast as they consume them. If you are confident that the consumers in your messaging system are that fast, consider setting the window size to -1. Setting the window size to this value allows unbounded message buffering on the client. Use this setting with caution, however. Memory on the client can overflow if the consumer is not able to process messages as fast as it receives them.

Setting the window size for fast consumers

The examples below show how to set the window size to -1 when using a AMQ Core Protocol JMS client that is a fast consumer of messages.

Procedure

  • If the client uses JNDI to instantiate its connection factory, include the consumerWindowSize parameter as part of the connection string URL. Store the URL within a JNDI context environment. The example below uses a jndi.properties file to store the URL. 

    java.naming.factory.initial=org.apache.activemq.artemis.jndi.ActiveMQInitialContextFactory
connectionFactory.myConnectionFactory=tcp://localhost:61616?consumerWindowSize=-1

  • If the client does not use JNDI to instantiate its connection factory, pass a value to ActiveMQConnectionFactory.setConsumerWindowSize(). 

    ConnectionFactory cf =  ActiveMQJMSClient.createConnectionFactory(...)
cf.setConsumerWindowSize(-1);

8.4. Handling slow consumers

Slow consumers take significant time to process each message. In these cases, buffering messages on the client is not recommended. Messages remain on the broker ready to be consumed by other consumers instead. One benefit of turning off the buffer is that it provides deterministic distribution between multiple consumers on a queue. To handle slow consumers by disabling the client-side buffer, set the window size to 0.

Setting the window size for slow consumers

The examples below show how to set the window size to 0 when using a AMQ Core Protocol JMS client that is a slow consumer of messages.

Procedure

  • If the client uses JNDI to instantiate its connection factory, include the consumerWindowSize parameter as part of the connection string URL. Store the URL within a JNDI context environment. The example below uses a jndi.properties file to store the URL. 

    java.naming.factory.initial=org.apache.activemq.artemis.jndi.ActiveMQInitialContextFactory
connectionFactory.myConnectionFactory=tcp://localhost:61616?consumerWindowSize=0

  • If the client does not use JNDI to instantiate its connection factory, pass a value to ActiveMQConnectionFactory.setConsumerWindowSize(). 

    ConnectionFactory cf =  ActiveMQJMSClient.createConnectionFactory(...)
cf.setConsumerWindowSize(0);

Additional resources

See the example no-consumer-buffering in <install-dir>/examples/standard for an example that shows how to configure the broker to prevent consumer buffering when dealing with slow consumers.

8.5. Setting the rate of message consumption

You can regulate the rate at which a consumer can consume messages. Also known as throttling, regulating the rate of consumption ensures that a consumer never consumes messages at a rate faster than configuration allows.

Note

Rate-limited flow control can be used in conjunction with window-based flow control. Rate-limited flow control affects only how many messages a client can consume per second and not how many messages are in its buffer. With a slow rate limit and a high window-based limit, the internal buffer of the client fills up with messages quickly.

The rate must be a positive integer to enable this functionality and is the maximum desired message consumption rate specified in units of messages per second. Setting the rate to -1 disables rate-limited flow control. The default value is -1.

The examples below show a client that limits the rate of consuming messages to 10 messages per second.

Procedure

  • If the client uses JNDI to instantiate its connection factory, include the consumerMaxRate parameter as part of the connection string URL. Store the URL within a JNDI context environment. The example below uses a jndi.properties file to store the URL. 

    java.naming.factory.initial=org.apache.activemq.artemis.jndi.ActiveMQInitialContextFactory
java.naming.provider.url=tcp://localhost:61616?consumerMaxRate=10

  • If the client does not use JNDI to instantiate its connection factory, pass the value to ActiveMQConnectionFactory.setConsumerMaxRate(). 

    ConnectionFactory cf =  ActiveMQJMSClient.createConnectionFactory(...)
cf.setConsumerMaxRate(10);

Additional resources

See the consumer-rate-limit example in <install-dir>/examples/standard for a working example of how to limit the consumer rate.

8.6. Setting the rate of message production

AMQ Core Protocol JMS can also limit the rate at which a producer sends messages. The producer rate is specified in units of messages per second. Setting it to -1, the default, disables rate-limited flow control.

The examples below show how to set the rate of sending messages when the producer is using AMQ Core Protocol JMS. Each example sets the maximum rate to 10 messages per second.

Procedure

  • If the client uses JNDI to instantiate its connection factory, include the producerMaxRate parameter as part of the connection string URL. Store the URL within a JNDI context environment. The example below uses a jndi.properties file to store the URL. 

    java.naming.factory.initial=org.apache.activemq.artemis.jndi.ActiveMQInitialContextFactory
java.naming.provider.url=tcp://localhost:61616?producerMaxRate=10

  • If the client does not use JNDI to instantiate its connection factory, pass the value to ActiveMQConnectionFactory.setProducerMaxRate(). 

    ConnectionFactory cf =  ActiveMQJMSClient.createConnectionFactory(...)
cf.setProducerMaxRate(10);

Additional resources

See the producer-rate-limit example in <install-dir>/examples/standard for a working example of how to limit a the rate of sending messages.