Concurrent processing of messages with strict order

Question

Within my JavaEE web application, I need to process incoming messages strictly in the order of arrival. I assume that my webapp container (Tomcat 6) retains the order of messages as they arrive at the http port.

What causes headaches to me, is the way I am internally processing these messages. For improved workload, I append the processing of each message to a ThreadPool as many things need to be done here e.g. XML parsing, sometimes enrichment of data using external web services. After processing is finished, I push the java representation of the message into a complex stream processing engine esper.codehaus.org, which is threadsafe. Here, different patterns are checked where the entrance order is the highest requirement e.g threshold of a phenomenon exceeds.

I had the idea of inserting every processed message into a PriorityQueue with a priority ID they received at time of arrival (in my Servlet, where it is incremented for each message). The problem is the following:

The thread which is polling the elements from the queue (lowest ID is the head of the queue) for inserting it into esper could skip an ID as it does not check for missing items. I guess an illustration works out better:

For steps (1) to (4) everything works as intended. But at step (5) the QueuePoller retrieves element 6 and not element 4 (which is later inserted at step (6)). This results in the message order: 2; 3; 6; 4.

What I tried to do was to alter the implementation of polling the queue's head to follow the strict order of IDs. Meaning, if the next ID's element is not yet inserted into the queue, wait at a barrier until its there. This seemed to work for the first 10 minutes but then hanged, probably due to an element that never got inserted into the queue.

Anyone out there having a similar problem in the past and has some hint for me?

Answer 1

Check out Disruptor -- a high performance queue with strict order (first enter -- first served)

Answer 2

You could instantly add a placeholder for incoming requests to your processing queue. The placeholder is preprocessed in the background by a thread pool but main processing waits for the preprocessing to be complete. The construct I have in mind is a Future.