Simple Java Messaging

Following up on my recent post Java Distributed Lock Manager, sometimes you just need a simple way to pass messages between Java processes.

Messaging is a very useful pattern in Enterprise Integration, and there are many ways to do it. Apache Camel is a great tool when you need the flexibility and power to manage complex messaging patterns, including routing, filtering and the like.

If you just want to do something simple, though, that can be a challenge. The most common solution, JMS, requires quite a bit of boilerplate code, and requires selecting and running a JMS provider, which means selecting a J2EE container, Apache ActiveMQ, or others.

So what if you just want a drop-dead simple way of adding messaging to your application? Terracotta gives you that. (And also integrates well with other solutions, like Apache Camel if you need more power later on).

Simple messaging in Terracotta is built on the notion of clustering a LinkedBlockingQueue. Just as a LinkedBlockingQueue is used to pass messages between threads in a single JVM, it will be used in combination with Terracotta's JVM-level clustering to provide message passing between JVMs.

To demonstrate, here is a simple example.

import java.io.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.*;

public class SimpleMessage
{
    private static ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
    private static BlockingQueue<String> queue = new LinkedBlockingQueue<String>();

    public static void receive() throws InterruptedException
    {
        System.out.println("Receiving messages...");
        while (true) {
            String msg = queue.take();
            System.out.println("msg >> " + msg);
        }
    }

    public static void send() throws Exception
    {
        while (true) {
            System.out.print("Enter a message> "); System.out.flush();
            String msg = new BufferedReader(new InputStreamReader(System.in)).readLine();
            queue.put(msg);
        }
   }

   public static void main(String[] args) throws Exception
    {
        // we use the presence of a lock to distinguish receiver from sender
        if (lock.writeLock().tryLock()) {
            receive();
        } else {
            send();
        }
    }
}

The app consist of two modes - a receiver mode and a sender mode. Normally, you would have an application specific mechanism of choosing whether you wanted to send messages or receive messages. For this example, we use a simple lock (for more information on using a ReentrantReadWriteLock with Terracotta, read the ReentrantReadWriteLock recipe). When free, the lock indicates no processes are receiving messages, so the process takes on the "receiver" mode. All subsequent processes take on the "sender" mode when the lock is held.

So let's run it with Terracotta and see how it works. First, we need to "cluster" the app. We need the lock and queue objects to be the same cluster-wide, which in Terracotta is called a root. So our Terracotta configuration file looks like:

<tc:tc-config xmlns:tc="http://www.terracotta.org/config"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://www.terracotta.org/schema/terracotta-4.xsd">

  <application>
    <dso>
      <roots>
        <root>
          <field-name>SimpleMessage.lock</field-name>
        </root>
        <root>
          <field-name>SimpleMessage.queue</field-name>
        </root>
      </roots>
    </dso>
  </application>
</tc:tc-config>

Now, let's run two JVMs with Terracotta. First, we start a server instance:

$ start-tc-server.sh
2008-12-14 10:26:18,246 INFO - Terracotta Server has started up as ACTIVE node on
  0.0.0.0:9510 successfully, and is now ready for work.

Then, we start our JVMs.

JVM 1:

$ dso-java.sh SimpleMessage
Receiving Messages...

JVM 2:

$ dso-java.sh SimpleMessage
Enter a message> 

Here, we enter a message, and see that it is printed in JVM 1:

JVM 2:

$ dso-java.sh SimpleMessage
Enter a message> hello world

JVM 1:

$ dso-java.sh SimpleMessage
Receiving Messages...
msg >> hello world

Further exploration

Try starting another JVM and see that they can both send messages to JVM 1. Try killing the receiver JVM and send messages to it. Then start another JVM. Since the lock is no longer held (Terracotta automatically releases any locks held by a JVM that exits the cluster) the new JVM will take on the receiver mode. Any messages sent while there was no receiver will have been queued, and will be printed on the startup of this new node.

And of course, you can see all the activity in the cluster. Try taking the receiver down again, send some messages using the sender nodes, then run the admin console. You'll be able to inspect the messages in the queue using the clustered heap browser.

This is just a demonstration of course - so to keep it simple I used a String as the message - but you could use any class.

For more fun with Terracotta, try the helpful "recipes" at Terracotta.org.

(Note, I've blogged about simple coordination in the past using Terracotta, which is similar)

Java Distributed Lock Manager

Sometimes you just need a simple way to coordinate activities across more than one java process. There's a lot of choices out there. The database, JMX, distributed caches, JMS, filesystems. It would be nice if there was a simple, easy way to get distributed locks in a J2SE, J2EE, Web, SOAP, or AJAX application? There is.

Terracotta provides one of the easiest ways to get a distributed lock manager in your Java application. Terracotta plugs right in to normal Java threading constructs—synchronized, wait/notify, java.concurrent.locks.ReentrantReadWriteLock, and even java.concurrent.CyclicBarrier, which means you basically already know how to use Terracotta as a lock manager.

To demonstrate, let's work up a simple locking example and then drop Terracotta in. Our app will consist of acquiring a lock, "do some work" in a simple loop, and repeat. Here's the code (LockExample.java):

public class LockExample
{
    private static Object lock = new Object();

    public static void main(String[] args) throws Exception
    {
        while (true) {
            System.out.print("Waiting for the lock..."); System.out.flush();
            synchronized (lock) {
                System.out.print("I got the lock, doing work");
                for (int i = 0; i < 4; i++) {
                    Thread.currentThread().sleep(1000);
                    System.out.print("."); System.out.flush();
                }
            }
            System.out.println("done");
       }
   }
}

Simple. If we run this on the command line, we get:

$ javac LockExample.java
$ java LockExample
Waiting for the lock...I got the lock, doing work....done
Waiting for the lock...I got the lock, doing work....done

During the "work" part the "."'s are added 1 every second for four seconds. Fancy.

Let's add Terracotta. We need a tc-config.xml file which tells Terracotta how to provide the appropriate clustering behavior to our application:

<tc:tc-config xmlns:tc="http://www.terracotta.org/config"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://www.terracotta.org/schema/terracotta-4.xsd">

  <application>
    <dso>
      <locks>
        <autolock>
           <method-expression>void LockExample.main(..)</method-expression>
        </autolock>
      </locks>
      <roots>
        <root>
          <field-name>LockExample.lock</field-name>
        </root>
      </roots>
    </dso>
  </application>
</tc:tc-config>

Now, let's run two JVMs with Terracotta. First, we start a server instance:

$ start-tc-server.sh
2008-12-11 22:26:18,246 INFO - Terracotta Server has started up as ACTIVE node on
  0.0.0.0:9510 successfully, and is now ready for work.

Then, we start our JVMs.

JVM 1:

$ dso-java.sh LockExample
Waiting for the lock...I got the lock, doing work....done

JVM 2:

$ dso-java.sh LockExample
Waiting for the lock...

It's a bit hard to demonstrate in a blog post, but the lock ping-pongs between the JVMs. That's it!

For more fun with distributed lock coordination, try these helpful "recipes":

• ReentrantReadWriteLock


• CyclicBarrier