Paging Example

This example shows how ActiveMQ would avoid running out of memory resources by paging messages.

A maxSize can be specified per Destination via the destinations settings configuration file (activemq-configuration.xml).

When messages routed to an address exceed the specified maxSize the server will begin to write messages to the file system, this is called paging. This will continue to occur until messages have been delivered to consumers and subsequently acknowledged freeing up memory. Messages will then be read from the file system , i.e. depaged, and routed as normal.

Acknowledgement plays an important factor on paging as messages will stay on the file system until the memory is released so it is important to make sure that the client acknowledges its messages.

Example step-by-step

To run the example, simply type mvn verify from this directory

1. First we need to get an initial context so we can look-up the JMS connection factory and destination objects from JNDI. This initial context will get it's properties from the client-jndi.properties file in the directory ../common/config

           InitialContext initialContext = getContext();
        

2. We look-up the JMS connection factory object from JNDI

           ConnectionFactory cf = (ConnectionFactory) initialContext.lookup("/ConnectionFactory");
        

3. We look-up the JMS queue object from JNDI. pagingQueue is configured to hold a very limited number of bytes in memory

           Queue pageQueue = (Queue) initialContext.lookup("/queue/pagingQueue");
        

4. We look-up the JMS queue object from JNDI.

           Queue queue = (Queue) initialContext.lookup("/queue/exampleQueue");
        

5. We create a JMS connection

           connection = cf.createConnection();
        

6. We create a JMS session. The session is created as non transacted. We will use client acknowledgement on this example.

           Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
        

7. Create a JMS Message Producer for pageQueueAddress

         MessageProducer pageMessageProducer = session.createProducer(pageQueue);
         

8. We don't need persistent messages in order to use paging. (This step is optional)

         pageMessageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
         

9. Create a Binary Bytes Message with 10K arbitrary bytes

         BytesMessage message = session.createBytesMessage();
         message.writeBytes(new byte[10 * 1024]);
         

10. Send only 20 messages to the Queue. This will be already enough for pagingQueue. Look at ./paging/config/activemq-queues.xml for the config.

         for (int i = 0; i < 20; i++)
         {
            pageMessageProducer.send(message);
         }         
         

11. Create a JMS Message Producer

         MessageProducer messageProducer = session.createProducer(queue);
         

12. We don't need persistent messages in order to use paging. (This step is optional)

         messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
         

13. Send the message for about 30K, which should be over the memory limit imposed by the server

         for (int i = 0; i < 30000; i++)
         {
            messageProducer.send(message);
         }
         

14. if you pause the example here, you will several files under ./build/data/paging

         // Thread.sleep(30000); // if you want to just our of curiosity, you can sleep here and inspect the created files just for 
         

15. Create a JMS Message Consumer

         MessageConsumer messageConsumer = session.createConsumer(queue);
         

16. Start the JMS Connection. This step will activate the subscribers to receive messages.

         connection.start();
         

17. Receive the messages. It's important to ACK for messages as ActiveMQ will not read messages from paging until messages are ACKed

         for (int i = 0; i < 30000; i++)
         {
            message = (BytesMessage)messageConsumer.receive(1000);

            if (i % 1000 == 0)
            {
               System.out.println("Received " + i + " messages");
               
               message.acknowledge();
            }
         }
         

18. Receive the messages from the Queue names pageQueue. Create the proper consumer for that.

         messageConsumer.close();
         messageConsumer = session.createConsumer(pageQueue);

         for (int i = 0; i < 20; i++)
         {
            message = (BytesMessage)messageConsumer.receive(1000);
            
            System.out.println("Received message " + i + " from pageQueue");

            message.acknowledge();
         }
         

19. And finally, always remember to close your JMS connections and resources after use, in a finally block. Closing a JMS connection will automatically close all of its sessions, consumers, producer and browser objects

           finally
           {
              if (initialContext != null)
              {
                initialContext.close();
              }
              if (connection != null)
              {
                 connection.close();
              }
           }