By default, ActiveMQ consumers buffer messages from the server in a client side buffer before actual delivery actually occurs.
This improves performance since otherwise every time you called receive() or had processed the last message in a MessageListener onMessage() method, the ActiveMQ client would have to go the server to request the next message involving a network round trip for every message reducing performance.
Therefore, by default, ActiveMQ pre-fetches messages into a buffer on each consumer. The total maximum size of
messages in bytes that will be buffered on each consumer is determined by the consumer-window-size
parameter on the connection factory.
In some cases it is not desirable to buffer any messages on the client side consumer.
An example would be an order queue which had multiple consumers that processed orders from the queue. Each order takes a significant time to process, but each one should be processed in a timely fashion.
If orders were buffered in each consumer, and a new consumer was added that consumer would not be able to process orders which were already in the client side buffer of another consumer.
To turn off client side buffering of messages, set consumer-window-size to zero.
With ActiveMQ you can specify a maximum consume rate at which a JMS MessageConsumer will consume messages.
This can be specified when creating or deploying the connection factory. See activemq-jms.xml
In this example we specify a consumer-window-size of 0 bytes in the activemq-jms.xml
file when deploying the connection factory:
<connection-factory name="ConnectionFactory">
<connector-ref connector-name="netty-connector"/>
<entries>
<entry name="ConnectionFactory"/>
</entries>
<!-- We set the consumer window size to 0, which means messages are not buffered at all
on the client side -->
<consumer-window-size>0</consumer-window-size>
</connection-factory>
We create a consumer on a queue and send 10 messages to it. We then create another consumer on the same queue.
We then consume messages from each consumer in a semi-random order. We note that the messages are consumed in the order they were sent.
If the messages had been buffered in each consumer they would not be available to be consumed in an order determined afer delivery.
To run the example, simply type mvn verify -Pexample from this directory
initialContext = getContext(0);
Queue queue = (Queue)initialContext.lookup("/queue/exampleQueue");
ConnectionFactory cf = (ConnectionFactory)initialContext.lookup("/ConnectionFactory");
connection = cf.createConnection();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
MessageProducer producer = session.createProducer(queue);
MessageConsumer consumer1 = session.createConsumer(queue);
connection.start();
final int numMessages = 10;
for (int i = 0; i < numMessages; i++)
{
TextMessage message = session.createTextMessage("This is text message: " + i);
producer.send(message);
}
MessageConsumer consumer2 = session.createConsumer(queue);
for (int i = 0; i < 3; i++)
{
TextMessage message = (TextMessage)consumer2.receive(2000);
System.out.println("Consumed message from consumer2: " + message.getText());
}
for (int i = 0; i < 5; i++)
{
TextMessage message = (TextMessage)consumer1.receive(2000);
System.out.println("Consumed message from consumer1: " + message.getText());
}
for (int i = 0; i < 2; i++)
{
TextMessage message = (TextMessage)consumer1.receive(2000);
System.out.println("Consumed message from consumer2: " + message.getText());
}
finally
{
if (initialContext != null)
{
initialContext.close();
}
if (connection != null)
{
connection.close();
}
}