Process multiple request from different threads in batches on one processing thread So I have a system that uses multiple threads to process data. These data could be processed individually but it would be better to process them in batches.

Lets assume we have a class Data, a class OtherData and a class Processor which implements Function&lt;List&lt;Data&gt;, List&lt;OtherData&gt;&gt;.

To process objects of type Data from multiple threads I designed two classes System and Evaluator.

public class System {
    private final Evaluator evalThread;
    private final Object sync = new Object();
    private Function&lt;List&lt;Data&gt;, List&lt;OtherData&gt;&gt; processor;
    private Map&lt;Object, Data&gt; inputMap;
    private Map&lt;Object, CompletableFuture&lt;OtherData&gt;&gt; futureMap;
    private List&lt;Object&gt; idList;

    public System() {
        processor = new Processor();
        inputMap = new HashMap&lt;&gt;();
        futureMap = new HashMap&lt;&gt;();
        idList = new LinkedList&lt;&gt;();

        evalThread = new Evaluator(processor, inputMap, futureMap, idList, sync);
        Thread thread = new Thread(evalThread, "EvalThread");
        thread.start();
    }

    public CompletableFuture&lt;OtherData&gt; process(Data data) {
        Object id = new Object();

        final CompletableFuture&lt;OtherData&gt; completableFuture = new CompletableFuture&lt;&gt;();

        synchronized (sync) {
            inputMap.put(id, data);
            futureMap.put(id, completableFuture);
            idList.add(id);

            if (idList.size() &gt;= 32) {
                sync.notifyAll();
            }
        }

        return completableFuture;
    }
}

public class Evaluator implements Runnable {
    private final Function&lt;List&lt;Data&gt;, List&lt;OtherData&gt;&gt; processor;
    private final Map&lt;Object, Data&gt; inputMap;
    private final Map&lt;Object, CompletableFuture&lt;OtherData&gt;&gt; futureMap;
    private final List&lt;Object&gt; idList;
    private final Object sync;

    private AtomicBoolean keepRunning = new AtomicBoolean(true);

    public Evaluator(Function&lt;List&lt;Data&gt;, List&lt;OtherData&gt;&gt; processor, Map&lt;Object, Data&gt; inputMap, Map&lt;Object,
                      CompletableFuture&lt;OtherData&gt;&gt; futureMap, List&lt;Object&gt; idList, Object sync) {
        this.processor = processor;
        this.inputMap = inputMap;
        this.futureMap = futureMap;
        this.idList = idList;
        this.sync = sync;
    }

    @Override
    public void run() {
        synchronized (sync) {
            while(keepRunning.get()) {
                if (idList.size() &gt; 0) {
                    List&lt;Data&gt; input = new LinkedList&lt;&gt;();

                    for (int i = 0; i &lt; idList.size(); i++) {
                        input.add(inputMap.get(idList.get(i)));
                    }

                    List&lt;OtherData&gt; output = processor.apply(input);

                    for (int i = 0; i &lt; idList.size(); i++) {
                        futureMap.get(idList.get(i)).complete(output.get(i));
                    }

                    idList.clear();
                    inputMap.clear();
                    futureMap.clear();
                }

                try {
                    sync.wait(20);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}



My idea was that any one can call process with singular data but the data will (if there are enough) be processed together with other Data objects.

Any suggestions for improvements or are there systems in the Java-framework that would fit this task better?
Do you might see problems according to deadlocks, etc.?
Well, technically you should really post complete code and not have
missing definitions.  Especially since Data and OtherData could
instead be generic type arguments perhaps, or interfaces, so that it's
actually clear what's happening.  I'm just gonna imagine they're
basically both Objects.  Also Processor is undefined.  If it was
really not important just pass it in as an argument to System, as it
is, this is incomplete too.



In System having a separate idList is pointless, the inputMap
already has the right size for implementing back pressure (I wanted to find a good definition, but right now at least Wikipedia doesn't have one under that term; basically if the input can grow unlimited, the system might not be able to catch up if processing takes too long and you might end up with a congested (read: out of memory) system).



In Evaluator, x.size() &gt; 0 could be !x.isEmpty(), that might
potentially be cheaper, but it also expresses intent a bit more clearly:
The check is really whether the map "isn't empty", not about how many
items are in the container exactly.

InterruptedException is for control flow,
don't just print a stack trace.

Also now that I read it, why's there three containers, inputMap,
futureMap and idList all related to the same task?  It'd be much
easier if there was just a sequence of tuples, Tuple&lt;Input, Future&gt;,
and then work through them.  Then replace the List with a
BlockingQueue&lt;Tuple&lt;Input, Future&gt;&gt; and it's already supporting the
waiting / back pressure too.  An ArrayBlockingQueue could be used to
limit the number of elements, while a LinkedBlockingQueue could have
an unbounded size (that's really not advisable though).

Lastly instead of keepRunning a tombstone object could then be
inserted to cancel the thread, then there's also no need for polling via
wait.  That is, insert a new Tuple&lt;&gt;() (same as the IDs at the
moment via new Object()) and when dequeuing from the input, check
whether this element was inserted, due to the object identity being
unique that's then safe to do.

So without spelling it all out, the main thread would look like this
perhaps:

public void run() {
    while (true) {
        Tuple&lt;Input, Future&gt; tuple = input.take();
        if (tuple == tombstone) {
            return;
        }
        tuple.getFuture().complete(processor.apply(tuple.getInput()));
    }
}


With some implementation of Tuple of course; a custom class would work
too.



For the other questions:  I can't see it deadlocking right now, the check for 32 and the notifyAll are a bit odd though.

Other tools in the Java tool set?  Yes, I'd suggest starting with reading through the java.util.concurrent namespace.  In particular the ExecutorService perhaps.