Chapter 19. Flow Control

JBoss EAP messaging includes configuration that defines how much data to pre-fetch on behalf of consumers and that controls the rate at which consumers can consume messages.

Window-based Flow Control

JBoss EAP messaging pre-fetches messages into a buffer on each consumer. The size of the buffer is determined by the consumer-window-size attribute of a connection-factory. The example configuration below shows a connection-factory with the consumer-window-size attribute explicitly set.

<connection-factory name="MyConnFactory" ... consumer-window-size="1048576" />

Use the management CLI to read and write the value of consumer-window-size attribute for a given connection-factory. The examples below show how this done using the InVmConnectionFactory connection factory, which is the default for consumers residing in the same virtual machine as the server, for example, a local MessageDrivenBean.

/subsystem=messaging-activemq/server=default/connection-factory=InVmConnectionFactory:read-attribute(name=consumer-window-size)
{
    "outcome" => "success",
    "result" => 1048576
}

/subsystem=messaging-activemq/server=default/connection-factory=InVmConnectionFactory:write-attribute(name=consumer-window-size,value=1048576)
{"outcome" => "success"}

The value for consumer-window-size must be an integer. Some values have special meaning as noted in the table below.

If you are using the core API, the consumer window size can be set from the ServerLocator using its setConsumerWindowSize() method.

If you are using JMS, the client can specify the consumer window size by using the setConsumerWindowSize() method of the instantiated ConnectionFactory.

Rate-limited Flow Control

JBoss EAP messaging can regulate the rate of messages consumed per second, a flow control method known as throttling. Use the consumer-max-rate attribute of the appropriate connection-factory to ensure that a consumer never consumes messages at a rate faster than specified.

<connection-factory name="MyConnFactory" ... consumer-max-rate="10" />

The default value is -1, which disables rate limited flow control.

The management CLI is the recommended way to read and write the consumer-max-rate attribute. The examples below show how this done using the InVmConnectionFactory connection factory, which is the default for consumers residing in the same virtual machine as the server, e.g. a local MessageDrivenBean.

/subsystem=messaging-activemq/server=default/connection-factory=InVmConnectionFactory:read-attribute(name=consumer-max-rate)
{
    "outcome" => "success",
    "result" => -1
}

/subsystem=messaging-activemq/server=default/connection-factory=InVmConnectionFactory:write-attribute(name=consumer-max-rate,value=100)
{"outcome" => "success"}

If you are using JMS the max rate size can be set using setConsumerMaxRate(int consumerMaxRate) method of the instantiated ConnectionFactory.

If you are using the Core API the rate can be set with the ServerLocator.setConsumerMaxRate(int consumerMaxRate) method.

JBoss EAP messaging can also limit the amount of data sent from a client in order to prevent the server from receiving too many messages.

Window-based Flow Control

JBoss EAP messaging regulates message producers by using an exchange of credits. Producers can send messages to an address as long as they have sufficient credits to do so. The amount of credits required to send a message is determined by its size. As producers run low on credits, they must request more from the server. Within the server configuration, the amount of credits a producer can request at one time is known as the producer-window-size, an attribute of the connection-factory element:

<connection-factory name="MyConnFactory" ... producer-window-size="1048576" />

The window size determines the amount of bytes that can be in-flight at any one time, thus preventing the remote connection from overloading the server.

Use the management CLI to read and write the producer-window-size attribute of a given connection factory. The examples below use the RemoteConnectionFactory, which is included in the default configuration and intended for use by remote clients.

subsystem=messaging-activemq/server=default/connection-factory=RemoteConnectionFactory:read-attribute(name=producer-window-size)
{
    "outcome" => "success",
    "result" => 65536
}

/subsystem=messaging-activemq/server=default/connection-factory=RemoteConnectionFactory:write-attribute(name=producer-window-size,value=65536)
{"outcome" => "success"}

If you are using JMS, the client can call the setProducerWindowSize(int producerWindowSize) method of the ConnectionFactory to set the window size directly.

If you are using the core API, the window size can be set using the setProducerWindowSize(int producerWindowSize) method of the ServerLocator.

Blocking Producer Window-based Flow Control

Typically, the messaging server always provides the same number of credits that was requested. However, it is possible to limit the number of credits sent by the server, which can prevent it from running out of memory due to producers sending more messages than can be handled at one time.

For example, if you have a JMS queue called myqueue and you set the maximum memory size to 10MB, the server will regulate the number of messages in the queue so that its size never exceeds 10MB. When the address gets full, producers will block on the client side until sufficient space is freed up on the address.

The address-setting configuration element contains the configuration for managing blocking producer flow control. An address-setting is used to apply a set of configuration to all queues registered to that address. See Configuring Address Settings for more information on how this is done.

For each address-setting requiring blocking producer flow control, you must include a value for the max-size-bytes attribute. The total memory for all queues bound to that address cannot exceed max-size-bytes. In the case of JMS topics, this means the total memory of all subscriptions in the topic cannot exceed max-size-bytes.

You must also set the address-full-policy attribute to BLOCK so the server knows that producers should be blocked if max-size-bytes is reached. Below is an example address-setting with both attributes set:

<address-setting ...
      name="myqueue"
      address-full-policy="BLOCK"
      max-size-bytes="100000" />

The above example would set the maximum size of the JMS queue "myqueue" to 100000 bytes. Producers will be blocked from sending to that address once it has reached its maximum size.

Use the management CLI to set these attributes, as in the examples below:

/subsystem=messaging-activemq/server=default/address-setting=myqueue:write-attribute(name=max-size-bytes,value=100000)
{"outcome" => "success"}

/subsystem=messaging-activemq/server=default/address-setting=myqueue:write-attribute(name=address-full-policy,value=BLOCK)
{"outcome" => "success"}

Rate-limited Flow Control

JBoss EAP messaging limits the number of messages a producer can send per second if you specify a producer-max-rate for the connection-factory it uses, as in the example below:

<connection-factory name="MyConnFactory" producer-max-rate="1000" />

The default value is -1, which disables rate limited flow control.

Use the management CLI to read and write the value for producer-max-rate. The examples below use the RemoteConnectionFactory, which is included in the default configuration and intended for use by remote clients.

/subsystem=messaging-activemq/server=default/connection-factory=RemoteConnectionFactory:read-attribute(name=producer-max-rate)
{
    "outcome" => "success",
    "result" => -1
}

/subsystem=messaging-activemq/server=default/connection-factory=RemoteConnectionFactory:write-attribute(name=producer-max-rate,value=100)
{"outcome" => "success"}

If you use the core API, set the rate by using the method ServerLocator.setProducerMaxRate(int producerMaxRate).

If you are using JNDI to instantiate and look up the connection factory, the max rate can be set on the client using the setProducerMaxRate(int producerMaxRate) method of the instantiated connection factory.