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 -Pexample 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
  2.            InitialContext initialContext = getContext();
            
  3. We look-up the JMS connection factory object from JNDI
  4.            ConnectionFactory cf = (ConnectionFactory) initialContext.lookup("/ConnectionFactory");
            
  5. We look-up the JMS queue object from JNDI. pagingQueue is configured to hold a very limited number of bytes in memory
  6.            Queue pageQueue = (Queue) initialContext.lookup("/queue/pagingQueue");
            
  7. We look-up the JMS queue object from JNDI.
  8.            Queue queue = (Queue) initialContext.lookup("/queue/exampleQueue");
            
  9. We create a JMS connection
  10.            connection = cf.createConnection();
            
  11. We create a JMS session. The session is created as non transacted. We will use client acknowledgement on this example.
  12.            Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
            
  13. Create a JMS Message Producer for pageQueueAddress
  14. 
             MessageProducer pageMessageProducer = session.createProducer(pageQueue);
             
  15. We don't need persistent messages in order to use paging. (This step is optional)
  16. 
             pageMessageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
             
  17. Create a Binary Bytes Message with 10K arbitrary bytes
  18. 
             BytesMessage message = session.createBytesMessage();
             message.writeBytes(new byte[10 * 1024]);
             
  19. Send only 20 messages to the Queue. This will be already enough for pagingQueue. Look at ./paging/config/activemq-queues.xml for the config.
  20. 
             for (int i = 0; i < 20; i++)
             {
                pageMessageProducer.send(message);
             }         
             
  21. Create a JMS Message Producer
  22. 
             MessageProducer messageProducer = session.createProducer(queue);
             
  23. We don't need persistent messages in order to use paging. (This step is optional)
  24. 
             messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
             
  25. Send the message for about 30K, which should be over the memory limit imposed by the server
  26. 
             for (int i = 0; i < 30000; i++)
             {
                messageProducer.send(message);
             }
             
  27. if you pause the example here, you will several files under ./build/data/paging
  28. 
             // Thread.sleep(30000); // if you want to just our of curiosity, you can sleep here and inspect the created files just for 
             
  29. Create a JMS Message Consumer
  30. 
             MessageConsumer messageConsumer = session.createConsumer(queue);
             
  31. Start the JMS Connection. This step will activate the subscribers to receive messages.
  32. 
             connection.start();
             
  33. Receive the messages. It's important to ACK for messages as ActiveMQ will not read messages from paging until messages are ACKed
  34. 
             for (int i = 0; i < 30000; i++)
             {
                message = (BytesMessage)messageConsumer.receive(1000);
    
                if (i % 1000 == 0)
                {
                   System.out.println("Received " + i + " messages");
                   
                   message.acknowledge();
                }
             }
             
  35. Receive the messages from the Queue names pageQueue. Create the proper consumer for that.
  36. 
             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();
             }
             
  37. 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
  38.            finally
               {
                  if (initialContext != null)
                  {
                    initialContext.close();
                  }
                  if (connection != null)
                  {
                     connection.close();
                  }
               }