Sending messages for processing

Question

I have a class which is responsible for waiting until a message is added to a message list and then sending it off to get processed.

• withdrawMessages waits for a message. I wait for a total of 2 minutes for a message to appear. I do this to keep the connection alive on my browser.
• depositMessage adds a message to the message list.

The process flow I am aiming for is:

• Deposit a message into the list

• Wait is notified and the whole list gets passed to the processor object (IUpdater)

If processMessages is about to get executed and a new message is added to the list I want to make sure this new message is also sent when calling responseUpdater.sendResponse

I need it to be thread-safe so when responseUpdater.sendResponse(messageList); is actually getting processed, then nothing else can be added to the list. It needs to wait until sendResponse is completed before adding it.

sendResponse basically sends the list to browser then the browser sends back a request to call withdrawMessages. So if a new message is added when sendResponse is getting executed, it has to wait until withdrawMessages is called again. At that point it will process the messages.

public class MessagePusher {

    private static final int TWO_MINUTE_WAIT = 120 * 1000;
    private volatile boolean notified = false;
    private final ArrayList<String> messageList = new ArrayList<String>();

    private void waitOnMessages(int duration) throws InterruptedException {
        messageList.wait(duration);
    }

    private void notifyWaitingThreads() {
        if (messageList.size() > 0) {
            notified = true;
            this.messageList.notify();
        }
    }

    public void depositMessage(String message) {
        synchronized (this.messageList) {
            this.messageList.add(message);
            notifyWaitingThreads();
        }
    }

    //IUpdater just processes the messages
    public void withdrawMessages(IUpdater updater) {
        boolean processRequired = false;
        synchronized (messageList) {
            //gets lock
            boolean timeout = false;
            try {
                //releases lock
                waitOnMessages(TWO_MINUTE_WAIT);
                //gets lock again
                if ((!notified)) {
                    timeout = true;
                }
            } catch (InterruptedException e) {
                ShipsLog.out.asWarning("Unexpected exception! Nested exception is:\n" + e);
            }
            if (notified || timeout) {
                notified = false;
                processRequired = true;
            }
        }
        if (processRequired) {
            processMessages(updater);
        }
    }

    public void processMessages(IUpdater responseUpdater) {
            synchronized (messageList) {
                responseUpdater.sendResponse(messageList);
                messageList.clear();
            }
    }
}

Answer 1

1. It looks like you only have one lock, so you can't have deadlock; but for future reference, you can add deadlock detection to the program, see for example this blog post but there are many other options available.

2. You may want to replace the arraylist with a ConcurrentLinkedQueue, this will let you remove the synchronized block from depositMessage and may let you remove the synchronized block from withdrawMessages and processMessages depending on how responseUpdater.sendReponse is implemented. In general, try to use lock-free data structures when possible - they generally scale better, and you can't heave deadlock if you don't have any locks.

3. ConTest is a tool for detecting multithreading problems e.g. race conditions. Java Pathfinder is a much more heavyweight approach to this.

4. For a large project, you may want to use a library like Akka or JMS to manage concurrency.