Concurrent in-order processing of work items from a Java BlockingQueue

user_214010 0
Viewed 207

I have part of a system that processes a BlockingQueue of input items within a worker thread, and puts the results on an BlockingQueue of output items, where the relevant code (simplified) looks something like this:

while (running()) {
   InputObject a=inputQueue.take();   // Get from input BlockingQueue
   OutputObject b=doProcessing(a);    // Process the item 
   outputQueue.put(b);                // Place on output BlockingQueue
}

doProcessing is the main performance bottleneck in this code, but the processing of queue items could be parallelised since the processing steps are all independent of each other.

I would therefore like to improve this so that items can be processed concurrently by multiple threads, with the constraint that this must not change the order of outputs (e.g. I can't simply have 10 threads running the loop above, because that might result in outputs being ordered differently depending on processing times).

What is the best way to achieve this in pure, idiomatic Java?

concurrency queue java performance
2 Answers

Parallel streams from List preserve ordering:

List<T> input = ...
List<T> output = input.parallelStream()
                .filter(this::running)
                .map(this::doProcessing)
                .collect(Collectors.toList());

PriorityBlockingQueue can be used if your work items can be compared to one another, and you will wait until running() is false before reading from the output queue:

outputQueue = new PriorityBlockingQueue<>();

Or you could order them after they have all been processed (if they can be compared to one another):

outputQueue.drainTo(outputList);
outputList.sort(null);

A simple way to implement comparation would be assigning a progressive ID to each element put into the input queue.

user_34532260

Create X event-loop threads, where X is the amount of steps that can be processed in parallel.

They will be processed in parallel, except one after another, i.e. not on the same item. While one step will be carried on on one item, the previous step will be carried on on the previous item, etc.

To further optimize it, you can use concurrent queues provided by JCTools, which are optimized for Single-Producer Single-Consumer scenarios (JDK's BlockingQueue implementations support Multiple-Producer Multiple-Consumer).

// Thread 1
while (running()) {
    InputObject a = inputQueue.take();
    OutputObject b = doProcessingStep1(a);
    queue1.put(b);
}

// Thread 2
while (running()) {
    InputObject a = queue1.take();
    OutputObject b = doProcessingStep2(a);
    queue2.put(b);
}

// Thread 3
while (running()) {
    InputObject a = queue2.take();
    OutputObject b = doProcessingStep3(a);
    outputQueue.put(b);
}
user_34532260
Related