Parallelizing deserialization step

There is the following pipeline:

1. item is produced (the producer is external to the pipeline);
2. item is deserialized (JSON to Java object);
3. item is processed;

At the moment it all happens synchronously in a single thread:

while(producer.next()) {
   var item = gson.deserialize(producer.item());
   
   processItem(item);
}

Or schematically:

PRODUCER -> DESERIALIZATION -> CONSUMER

(sync)      (sync)            (sync)

The concern is that the deserialization step has no side-effects and could be parallelized saving some world time.

The overall code should like the following:

var pipeline = new Pipeline<Item>();
pipeline.setProducer(producer);
pipeline.setDeserialization(gson::deserialize);
pipeline.setConsumer(item -> {
    ...
});
pipeline.run();

Or schematically:

         -> DESERIALIZATION
         -> DESERIALIZATION
         -> DESERIALIZATION
PRODUCER -> ...             -> CONSUMER
         -> DESERIALIZATION
         -> DESERIALIZATION
         -> DESERIALIZATION

(sync)      (parallel)         (sync)

Important notice. Deserialized items should be produced:

• synchronously;
• in the same order the original producer produces encoded items.

Q. Is there a standardized way to code such a pipeline?

Try

while(producer.next()) {
   CompletableFuture.supplyAsync(()-> gson.deserialize(producer.item()))
    .thenRunAsync(item->processItem(item));
}

One way you can achieve your pattern is to:

• Construct a multi-threaded executor to process the decoding requests
• Have a consumer queue; each time you submit an item to be decoded, also add the corresponding Future object to the consumer queue
• Have a consumer thread sit waiting to take items off the queue [which therefore consumes them in the order they were posted], call the corresponding get() method [which waits for the item to be decoded]

So the 'consumer' would look like this:

BlockingQueue<Future<Item>> consumerQueue = new LinkedBlockingDeque<>();
Thread consumerThread = new Thread(() -> {
        try {
            while (true) {
                Future<Item> item = consumerQueue.take();
                try {
                    // Get the next decoded item that's ready
                    Item decodedItem = item.get();
                    // 'Consume' the item
                    ...
                } catch (ExecutionException ex) {
                }
            }
        } catch (InterruptedException irr) {

        }
    });
    consumerThread.start()

Meanwhile, the 'producer' end, with its multi-threaded 'decoder', would look like this:

ExecutorService decoder = Executors.newFixedThreadPool(4);
while (!producer.hasNext()) {
        Item item = producer.next()

        // Submit the decode job for asynchronous processing
        Future<Item> p = decoder.submit(() -> {
            item.decode();
        }, item);

        // Also queue this decode job for future consumption once complete
        consumerQueue.add(p);
    }

As a separate matter, I wonder if you will actually see much benefit in practice, since by insisting on consumption in the same order, you are inherently introducing a serial condition on the process. But technically, this is one way that you could achieve what you are after.

P.S. If you didn't want a separate consumer thread, then the same 'producer' thread could poll the queue for completed items and execute in line.