I can have a maximum of three concurrent threads processing tasks. Each of these threads can process 1 to 100 tasks simultaneously (by connecting to an external system). The time taken to process 100 tasks at once in one thread is the same as it takes to process 1 task in one thread because the overhead of connecting to the external system is what takes 95% of the time. Tasks come in at random intervals from other threads in the application, these threads need to block until the task is done, or a timeout is hit. Responding quickly to the threads submitting tasks is the primary goal here, and making use of the ability to process tasks in batch is secondary (but useful and important when we have a big queue of tasks as it saves time).This code works (or appears to anyway), but before I build it into my larger application I want to check that this can't be done better/faster etc. It has the potential to create a major bottleneck in the application so I want to make sure it is done in the most efficient way possible. I'm not very experienced with the concurrent package. Any feedback would be greatly appreciated.Here is the concept code that I have written to test with. System outs will be replaced with logging when it is built into the main application.import java.util.ArrayList;import java.util.List;import java.util.concurrent.Executor;import java.util.concurrent.Executors;import java.util.concurrent.LinkedBlockingQueue;import java.util.concurrent.Semaphore;public class TaskProcessor implements Runnable{    private final LinkedBlockingQueue<Task> queue = new LinkedBlockingQueue<>();    private final Semaphore semaphore = new Semaphore(3);    private final Executor executor = Executors.newCachedThreadPool();    public static void main(final String[] args)    {        final TaskProcessor taskProcessor = new TaskProcessor();        new Thread(taskProcessor).start();        for (int i = 0; i < 20; i++)        {            new Thread(new Runnable()            {                @Override                public void run()                {                    try                    {                        taskProcessor.submitTask(new Task());                    }                    catch (final InterruptedException e)                    {                        throw new RuntimeException(e); // unreachable                    }                }            }).start();        }    }    private static class Task    {        public void setResult() {}        public void getResult() {}    }    public void submitTask(final Task task) throws InterruptedException    {        final long taskStartTime = System.currentTimeMillis();        this.queue.add(task);        synchronized (task)        {            task.wait();            System.out.println(task complete after  + (System.currentTimeMillis() - taskStartTime) + ms);        }        task.getResult();    }    @Override    public void run()    {        try        {            while (true)            {                this.semaphore.acquire(); // blocks until we can get a permit                final List<Task> tasks = new ArrayList<>();                System.out.println(waiting for task);                tasks.add(this.queue.take()); // will block if the queue is empty                // get all the other tasks in the queue                Task nextTask = this.queue.poll();                while (nextTask != null && tasks.size() < 100)                {                    tasks.add(nextTask);                    nextTask = this.queue.poll();                }                this.processTasks(tasks);            }        }        catch (final InterruptedException e)        {            throw new RuntimeException(e); // unreachable        }    }    private void processTasks(final List<Task> tasks)    {        final Runnable runnable = new Runnable()        {            @Override            public void run()            {                try                {                    TaskProcessor.this.executeTasks(tasks);                    for (final Task task : tasks)                    {                        task.setResult(); // once fully implemented will set the result to be the result of exeuteTasks()                        synchronized (task)                        {                            task.notify();                        }                    }                }                finally                {                    TaskProcessor.this.semaphore.release();                }            }        };        this.executor.execute(runnable);    }    private void executeTasks(final List<Task> tasks)    {        try        {            System.out.println(executing  + tasks.size() +  tasks.);            Thread.sleep(1000);        }        catch (final InterruptedException e)        {            e.printStackTrace();        }    }}
Concurrent task processing queue
java;concurrency