Handling messages from a server

Question

I've never really done anything with concurrency, so I'm not sure if this is done correctly.

Basically, I have a class that handles all messages from the server and decides what to do them. It uses a separate thread to keep looping through a list to check for any new messages it has to deal with.

The main server thread needs to be able to add messages to the ArrayList without causing a concurrency error, and so far it seems ok but I don't have that many messages coming in.

public class MessageThread implements Runnable {
    public static ArrayList<Message> messages = new ArrayList<Message>();

    @Override
    public void run() {
        while (true) {
            synchronized (messages) {

                for (Message m : messages) {

        }
    }
}
    public static void addMessage(Message message) {
        synchronized (messages) {
            messages.add(message);
        }
    }

    public static void removeMessage(Message message) {
        synchronized (messages) {
            messages.remove(message);
        }
    }
}

So as you can see, the server thread would use the MessageThread.addMessage method whenever it needs to add a message.

Answer 1

Your ArrayList is correctly guarded from access by multiple threads but you've got another problem. The busy wait in your run method is going to hog the CPU and might even starve the other threads that would be trying to put a message in the queue. You'll need some way of signaling that the queue is non-empty so that the run function can wait on the signal. In .NET it would be a ManualResetEvent. I don't know the java equivalent. Or better yet, have a look at BlockingQueue.

@Override
public void run() {
    while (true) {
        queueFullEvent.waitOne();
        synchronized (messages) {
            for (Message m : messages) {
                // messages should be removed here
                messages.remove(m);
                if (messages.isEmpty)
                    queueFullEvent.reset();    
            }
        }
    }
}

public static void addMessage(Message message) {
    synchronized (messages) {
        messages.add(message);
        queueFullEvent.set();
    }
}

public static void removeMessage(Message message) {
   // would require additional syncronization objects to avoid a race.
}

}

Answer 2

1. public static ArrayList<Message> messages .... It is classic confined object escape. It means that since your synchronization object is public, somebody else can acquire its monitor outside MessageThread class, and cause a deadlock. Use proper encapsulation for your intrinsic locks.
2. It is not clear why public static void removeMessage(Message message) is a part of your public API. Am I understand right that the server thread is supposed to call MessageThread.addMessage first, then somehow it must figure out that the message has been already processed, and then call MessageThread.removeMessage ? Probably, it should be a MessageThread's responsibility to clean up completed messages from the underlying data structure ?
3. There is an existing Java API for such use case. Please, take a look at java.util.concurrent.ArrayBlockingQueue class. Also consider using java.util.concurrent.Executors to process messages asynchronously, if needed.

Answer 3

Java synchronizes on objects (and not references), as such it is always good to make any reference that you will later use for synchronization final. so

public static ArrayList<Message> messages = new ArrayList<Message>();

should be

public static final ArrayList<Message> messages = new ArrayList<Message>();

This is because if in one your synchronized code blocks you do messages = anotherListOfMessages then other code blocks waiting on the lock might think it has been released when actually it (the reference) is now pointing to another object. The behavior that follows will most likely be something you don't expect.

Answer 4

I'm surprised no one has mentioned this yet, but before going to far down the road you should really check out Java Concurrency In Practice. It is by far the best resource for learning about concurrency on the JVM. It will cover the fundamental concurrency problems, idiosyncracies of the JVM platform and all the utility classes available.

As for your implementation specifically, there are a number of suspicious things in this class... First of all there is probably no need for the internal message queue to be public. The number one source of concurrency errors is shared mutable state. So the less you share and the less that is mutable the better off you will be.

While it doesn't directly affect the concurrency, is there a reason for choosing to make everything static? Generally static-ness can make it harder to reason about concurrency issues, so unless there is a compelling reason to do so I would make this class not static and then inject/provide instances where needed.

Finally, it has been mentioned but Java has a suite of concurrency related utilities and classes in the java.util.concurrent package. Concurrency, like security is best left to the "professionals" and until you are one the best advice is to stick to the util classes provided and avoid making your own clever concurrency designs.