Large Message Example

This example shows you how to send and receive very large messages with ActiveMQ.

ActiveMQ supports the sending and receiving of huge messages, much larger than can fit in available RAM on the client or server. Effectively the only limit to message size is the amount of disk space you have on the server.

Large messages are persisted on the server so they can survive a server restart. In other words ActiveMQ doesn't just do a simple socket stream from the sender to the consumer.

In order to do this ActiveMQ provides an extension to JMS where you can use an InputStream or OutputStream as the source and destination for your messages. You can send messages as large as it would fit in your disk.

You may also choose to read LargeMessages using the regular ByteStream or ByteMessage methods, but using the InputStream and OutputStream will provide you a much better performance

Example step-by-step

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

In this example we limit both the server and the client to be running in a maximum of 50MB of RAM, and we send a message with a body of size 256MB.

ActiveMQ can support much large message sizes but we choose these sizes and limit RAM so the example runs more quickly.

We create a file on disk representing the message body, create a FileInputStream on that file and set that InputStream as the body of the message before sending.

The message is sent, then we stop the server, and restart it. This demonstrates the large message will survive a restart of the server.

Once the server is restarted we receive the message and stream it's body to another file on disk.

Create an initial context to perform the JNDI lookup.

           initialContext = getContext(0);

Perfom a lookup on the queue.

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

Perform a lookup on the Connection Factory. This ConnectionFactory has a special attribute set on it. activemq-jms.xml) Messages with more than 10K are considered large.

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

Create the JMS objects for sending the message.

           
        connection = cf.createConnection();

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

        MessageProducer producer = session.createProducer(queue);

Create a huge file - this will form the body of the message we will send.

           
        File fileInput = new File("huge_message_to_send.dat");

        fileInput.createNewFile();

        createFile(fileInput, FILE_SIZE);

Create a BytesMessage

          BytesMessage message = session.createBytesMessage();

We set the InputStream on the message. When sending the message will read the InputStream until it gets EOF. In this case we point the InputStream at a file on disk, and it will suck up the entire file, however we could use any InputStream not just a FileInputStream.


         FileInputStream fileInputStream = new FileInputStream(fileInput);
         
         BufferedInputStream bufferedInput = new BufferedInputStream(fileInputStream);
         
         message.setObjectProperty("JMS_HQ_InputStream", bufferedInput);

Send the Message.


         producer.send(message);

We send message to the queue. After the send completion the message file will be located at ./build/data/largeMessages

           messageProducer.send(message);

To demonstrate that that we're not simply streaming the message from sending to consumer, we stop the server and restart it before consuming the message. This demonstrates that the large message gets persisted, like a normal persistent message, on the server. If you look at ./build/data/largeMessages you will see the largeMessage stored on disk the server.

           
        connection.close();

        initialContext.close();

        stopServer(0);

        // Give the server a little time to shutdown properly
        Thread.sleep(5000);

        startServer(0);

Now the server is restarted we can recreate the JMS Objects, and start the new connection.

           
        initialContext = getContext(0);

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

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

        connection = cf.createConnection();

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

        MessageConsumer messageConsumer = session.createConsumer(queue);

        connection.start();

Receive the message. When we receive the large message we initially just receive the message with an empty body.

BytesMessage messageReceived = (BytesMessage)messageConsumer.receive(120000);

We set an OutputStream on the message. This causes the message body to be written to the OutputStream until there are no more bytes to be written. You don't have to use a FileOutputStream, you can use any OutputStream. You may choose to use the regular BytesMessage or StreamMessage interface but this method is much faster for large messages.


         File outputFile = new File("huge_message_received.dat");

         FileOutputStream fileOutputStream = new FileOutputStream(outputFile);

         BufferedOutputStream bufferedOutput = new BufferedOutputStream(fileOutputStream);

This will save the stream and wait until the entire message is written before continuing.


         messageReceived.setObjectProperty("JMS_HQ_SaveStream", bufferedOutput);

Be sure to close our resources!

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