CODEWITHSUNDEEP
javaperformanceconcurrencyqueue
mikera
mikera

Reputation: 106381

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

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?

Upvotes: 2

Views: 334

Answers (2)

user3453226
user3453226

Reputation:

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.

Upvotes: 2

user3453226
user3453226

Reputation:

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);
}

Upvotes: 1

Related Questions