Is there an out-of-the-box thread pool with multiple queues (that ensure serial processing of each queue)?

java high performance thread pool
dynamic thread pool java
java thread pool
extend threadpoolexecutor example
how tomcat thread pool works
how to decide number of threads in threadpool
java task queue
stripedexecutor

Among all my tasks, I have some that must be processed serially (they can never run concurrently and they must be processed in order).

I achieved that creating a separated thread pool with a single thread for each group of tasks that must be executed serially. It works but I don't have the resources for that. I don't control the number of groups, so I might end up with a ridiculous number of threads running simultaneously.

Is there any way I can accomplish that with a single thread pool? Is there a thread pool with multiple blocking queues where I could ensure serial execution for each queue?

EDIT:

Just emphasizing what I've said in my second paragraph: I've solved this with a single threaded thread pool for each group of tasks that must be executed serially. I can't go on with this solution, though. There are way too many groups and I can't have all these threads.

I've found this related question, but since it is not very recent, I still created mine. All I'm doing is trying to avoid reinventing the wheel, but it seems I don't have a choice.

Does Java have an indexable multi-queue thread pool?

If you maintain a queue for each group, you can pull items off each queue and feed them into a thread pool. The code below won't prioritize any one group, it just pulls them in a round-robin fashion. If you need to add prioritization you should easily be able to. The following code will round-robin 4 groups using two threads (plus the thread managing the queue). You can use another queue mechanism. I typically use LinkedBlockingQueue for situations where I want to wait for items to be placed on the queue by another thread, which probably is not what you want - so I'm polling instead of calling take(). Take is the call that waits.

private Future group1Future = null;
private Future group2Future = null;
private Future group3Future = null;
private Future group4Future = null;
private LinkedBlockingQueue<Callable> group1Queue
        = new LinkedBlockingQueue<>();
private LinkedBlockingQueue<Callable> group2Queue
        = new LinkedBlockingQueue<>();
private LinkedBlockingQueue<Callable> group3Queue
        = new LinkedBlockingQueue<>();
private LinkedBlockingQueue<Callable> group4Queue
        = new LinkedBlockingQueue<>();

private ExecutorService executor = Executors.newFixedThreadPool(2);


public void startProcessing() {
    while (true) {
        if (group1Future != null && group1Future.isDone()) {
            if (group1Queue.peek() != null) {
                group1Future = executor.submit(group1Queue.poll());
            }
        }
        if (group2Future != null && group1Future.isDone()) {
            if (group2Queue.peek() != null) {
                group2Future = executor.submit(group2Queue.poll());
            }
        }
        if (group3Future != null && group3Future.isDone()) {
            if (group3Queue.peek() != null) {
                group3Future = executor.submit(group3Queue.poll());
            }
        }

        if (group4Future != null && group4Future.isDone()) {
            if (group4Queue.peek() != null) {
                group4Future = executor.submit(group4Queue.poll());
            }
        }
    }
}

If a task for that group is not complete, it will skip to the next group. No more than two groups will be processed at a time and no single group will ever run more than one task. The queues will enforce ordered execution.

java, with multiple queues (that ensure serial processing of each queue)? you can pull items off each queue and feed them into a thread pool. 18 Is there an out-of-the-box thread pool with multiple queues (that ensure serial processing of each queue)? View more network posts → Top tags (13)

Akka, as suggested by @SotiriosDelimanolis and @AlexeiKaigorodov seems promising, as well as @Dodd10x second answer, which certainly solves the problem. The only downside is that I'd have to code my own polling strategy to make sure my tasks are eventually added to the executor (like the infinite loop in his example).

On the other hand, the Striped Executor Service suggested by @OldCurmudgeon exactly matches my problem and works out of the box simply as a custom ExecutorService.

This magical thread pool would ensure that all Runnables with the same stripeClass would be executed in the order they were submitted, but StripedRunners with different stripedClasses could still execute independently. He wanted to use a relatively small thread pool to service a large number of Java NIO clients, but in such a way that the runnables would still be executed in-order.

There is even a comment about using a single threaded thread pool for each group (stripe), as it was suggested here:

Several suggestions were made, such as having a SingleThreadExecutor for each stripeClass. However, that would not satisfy the requirement that we could share the threads between connections.

I see this as the best solution for its simplicity and ease of use.

When using Task what happens if the ThreadPool is full/busy , How do I know how many threads My ThreadPool has? Download source - 9.88 KB; Introduction. A user on StackOverflow recently asked a question about processing items from a Queue<T> using multiple threads. While a general solution to this problem is to enqueue these operations using the .NET ThreadPool mechanism, often this sort of task can scale itself out of being a feasible candidate for using the ThreadPool fairly easily; too many

A single thread executor will do

ExecutorService  executorService = Executors.newSingleThreadExecutor();

Which internally uses a ThreadPoolExecutor with a LinkedBlockingQueue

new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()))

So you can use this for your sequential stuff and probably use a multi-threaded executor service for concurrent tasks

How to set an ideal thread pool size, of available cores. If all tasks are computation intensive, then this is all we need. Consumer thread will automatically wait until BlockingQueue is not populated with some data. Once it fills, the thread will consume the resource. BlockingQueue works on following rules: If fewer than corePoolSize threads are running, the Executor always prefers adding a new thread rather than queuing.

Look into Java's built-in thread executor service.

http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ExecutorService.html

There is a single thread executor that will process each task synchronously.

In response to the comments section:

Please read the API before you say this won't work. http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Executors.html#newSingleThreadExecutor()

public static ExecutorService newSingleThreadExecutor() Creates an Executor that uses a single worker thread operating off an unbounded queue. (Note however that if this single thread terminates due to a failure during execution prior to shutdown, a new one will take its place if needed to execute subsequent tasks.) Tasks are guaranteed to execute sequentially, and no more than one task will be active at any given time. Unlike the otherwise equivalent newFixedThreadPool(1) the returned executor is guaranteed not to be reconfigurable to use additional threads.

Note: is states they are guaranteed to execute sequentially.

EDIT:

Now that I understand your question better, I have an idea you could try. If you maintain a queue for each group, you can pull items off each queue and feed them into a thread pool. The code below won't prioritize any one group, it just pulls them in a round robbing fashion. If you need to add prioritization you should easily be able to. The following code will round robbing 4 groups using two threads (plus the thread managing the queue). You can use another queue mechanism. I typically use LinkedBlockingQueue for situations where I want to wait for items to be placed on the queue by another thread, which probably is not what you want - which is why I'm polling instead of calling take(). Take is the call that waits.

private Future group1Future = null;
private Future group2Future = null;
private Future group3Future = null;
private Future group4Future = null;
private LinkedBlockingQueue<Callable> group1Queue
        = new LinkedBlockingQueue<>();
private LinkedBlockingQueue<Callable> group2Queue
        = new LinkedBlockingQueue<>();
private LinkedBlockingQueue<Callable> group3Queue
        = new LinkedBlockingQueue<>();
private LinkedBlockingQueue<Callable> group4Queue
        = new LinkedBlockingQueue<>();

private ExecutorService executor = Executors.newFixedThreadPool(2);


public void startProcessing() {
    while (true) {
        if (group1Future != null && group1Future.isDone()) {
            if (group1Queue.peek() != null) {
                group1Future = executor.submit(group1Queue.poll());
            }
        }
        if (group2Future != null && group1Future.isDone()) {
            if (group2Queue.peek() != null) {
                group2Future = executor.submit(group2Queue.poll());
            }
        }
        if (group3Future != null && group3Future.isDone()) {
            if (group3Queue.peek() != null) {
                group3Future = executor.submit(group3Queue.poll());
            }
        }

        if (group4Future != null && group4Future.isDone()) {
            if (group4Queue.peek() != null) {
                group4Future = executor.submit(group4Queue.poll());
            }
        }
    }
}

If a task for that group is not complete, it will skip to the next group. No more than two groups will be processed at a time and no single group will ever run more than one task. The queues will enforce ordered execution.

A Deep Dive Into the Java ExecutorService, What is a thread pool and why is it used? There is a lot that needs to be taken into consideration when implementing a threaded work queue and I wouldn't recommend doing it manually unless you've got a really good reason because existing solutions are tested and reliable and without very specific design needs and some hardcore coding to go with it you'll likely wind up with something less efficient/powerful than the existing options.

I recently answered a question about a "serial task queue" with a basic implementation as demonstration here. I imagine you have been using a similar solution. It is relatively easy to adapt the implementation to use a map of task lists and still share one (fixed size) executor. The Striped Executor Service you mention is the better solution, but I show the adapted implementation here to demonstrate decoupling the task queue(s) from the executor. The implementation uses a callback and therefor has no need to do polling or signalling. Since a "critical (stop the world) section" is used, the map with task queues can clean itself: no tasks queued means empty map. Downside of the "critical section" is that throughput is limited: only so many tasks can be added and removed per second.

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;

// Copied and updated from https://stackoverflow.com/a/32916943/3080094
public class SerialTaskQueues {

    public static void main(String[] args) {

        // test the serial task execution using different groups
        ExecutorService executor = Executors.newFixedThreadPool(2);
        SerialTaskQueues tq = new SerialTaskQueues(executor);
        try {
            // test running the tasks one by one
            tq.add(new SleepSome("1", 30L));
            Thread.sleep(5L);
            tq.add(new SleepSome("2", 20L));
            tq.add(new SleepSome("1", 10L));

            Thread.sleep(100L);
            // all queues should be empty
            System.out.println("Queue size 1: " + tq.size("1")); // should be empty
            System.out.println("Queue size 2: " + tq.size("2")); // should be empty
            tq.add(new SleepSome("1", 10L));
            tq.add(new SleepSome("2", 20L));
            // with executor pool size set to 2, task 3 will have to wait for task 1 to complete
            tq.add(new SleepSome("3", 30L));
            tq.add(new SleepSome("1", 20L));
            tq.add(new SleepSome("2", 10L));

            Thread.sleep(100L);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            executor.shutdownNow();
        }
    }

    // all lookups and modifications to the list must be synchronized on the list.
    private final Map<String, GroupTasks> taskGroups = new HashMap<>();
    // make lock fair so that adding and removing tasks is balanced.
    private final ReentrantLock lock = new ReentrantLock(true);
    private final ExecutorService executor;

    public SerialTaskQueues(ExecutorService executor) {
        this.executor = executor;
    }

    public boolean add(String groupId, Runnable task) {

        lock.lock();
        try {
            GroupTasks gt = taskGroups.get(groupId);
            if (gt == null) {
                gt = new GroupTasks(groupId);
                taskGroups.put(groupId, gt);
            }
            gt.tasks.add(task); 
        } finally {
            lock.unlock();
        }
        runNextTask(groupId);
        return true;
    }

    /* Utility method for testing. */
    public void add(SleepSome sleepTask) {
        add(sleepTask.groupId, sleepTask);
    }

    private void runNextTask(String groupId) {

        // critical section that ensures one task is executed.
        lock.lock();
        try {
            GroupTasks gt = taskGroups.get(groupId);
            if (gt.tasks.isEmpty()) {
                // only cleanup when last task has executed, prevent memory leak
                if (!gt.taskRunning.get()) {
                    taskGroups.remove(groupId);
                }
            } else if (!executor.isShutdown() && gt.taskRunning.compareAndSet(false, true)) {
                executor.execute(wrapTask(groupId, gt.taskRunning, gt.tasks.remove(0)));
            }
        } finally {
            lock.unlock();
        }
    }

    private CallbackTask wrapTask(final String groupId, final AtomicBoolean taskRunning, Runnable task) {

        return new CallbackTask(task, new Runnable() {
            @Override 
            public void run() {
                if (!taskRunning.compareAndSet(true, false)) {
                    System.out.println("ERROR: programming error, the callback should always run in execute state.");
                }
                runNextTask(groupId);
            }
        });
    }

    /** Amount of (active) task groups. */
    public int size() {

        int size = 0;
        lock.lock();
        try {
            size = taskGroups.size();
        } finally {
            lock.unlock();
        }
        return size;
    }

    public int size(String groupId) {

        int size = 0;
        lock.lock();
        try {
            GroupTasks gt = taskGroups.get(groupId);
            size = (gt == null ? 0 : gt.tasks.size());
        } finally {
            lock.unlock();
        }
        return size;
    }

    public Runnable get(String groupId, int index) {

        Runnable r = null;
        lock.lock();
        try {
            GroupTasks gt = taskGroups.get(groupId);
            r =  (gt == null ? null : gt.tasks.get(index));
        } finally {
            lock.unlock();
        }
        return r;
    }

    public Runnable remove(String groupId, int index) {

        Runnable r = null;
        lock.lock();
        try {
            GroupTasks gt = taskGroups.get(groupId);
            r = gt.tasks.remove(index);
            // similar to runNextTask - cleanup if there are no tasks (running) for the group 
            if (gt.tasks.isEmpty() && !gt.taskRunning.get()) {
                taskGroups.remove(groupId);
            }
        } finally {
            lock.unlock();
        }
        return r;
    }

    /* Helper class for the task-group map. */
    class GroupTasks {

        final List<Runnable> tasks = new LinkedList<Runnable>();
        // atomic boolean used to ensure only 1 task is executed at any given time
        final AtomicBoolean taskRunning = new AtomicBoolean(false);
        final String groupId;

        GroupTasks(String groupId) {
            this.groupId = groupId;
        }
    }

    // general callback-task, see https://stackoverflow.com/a/826283/3080094
    static class CallbackTask implements Runnable {

        private final Runnable task, callback;

        public CallbackTask(Runnable task, Runnable callback) {
            this.task = task;
            this.callback = callback;
        }

        @Override 
        public void run() {

            try {
                task.run();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                try {
                    callback.run();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    // task that just sleeps for a while
    static class SleepSome implements Runnable {

        static long startTime = System.currentTimeMillis();

        private final String groupId;
        private final long sleepTimeMs;
        public SleepSome(String groupId, long sleepTimeMs) {
            this.groupId = groupId;
            this.sleepTimeMs = sleepTimeMs;
        }
        @Override public void run() {
            try { 
                System.out.println(tdelta(groupId) + "Sleeping for " + sleepTimeMs + " ms.");
                Thread.sleep(sleepTimeMs);
                System.out.println(tdelta(groupId) + "Slept for " + sleepTimeMs + " ms.");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        private String tdelta(String groupId) { return String.format("% 4d [%s] ", (System.currentTimeMillis() - startTime), groupId); }
    }
}

Thread Pools, Finally understanding how thread pools really work in Java can be the Multithreaded programming refers to using threads to execute multiple tasks concurrently. By repeating this process every time we need to execute a task, the Runnable task in a queue to wait for a thread to become available. Queues the specified delegate to the thread pool, but does not propagate the calling stack to the worker thread. UnsafeQueueUserWorkItem<TState>(Action<TState>, TState, Boolean) Queues a method specified by an Action<T> delegate for execution, and specifies an object containing data to be used by the method.

Finally Getting the Most out of the Java Thread Pool, Within a process or program, we can run multiple threads concurrently to improve the A thread must carve out its own resources within the running process. Channeling all accesses to GUI components in a single thread ensure thread safety. for (int i = 1; i <= 3; ++i) { System.out.println("Consumer thread 1 Get " + box. NUMA nodes are ignored. There will be just one IOProcess thread pool, and all threads in that thread pool will be affinitized to all logical processors. By default (where PerNumaNode=-1), this is the operative setting if the computer has fewer than 4 NUMA nodes. Setting PerNumaNode=1. IOProcess thread pools are created for each NUMA node.

Multithreading and Concurrency - Java Programming Tutorial, So, programs can create multiple threads, and all of those threads can run code and access the process's data simultaneously, which can result in faster  One of the most common questions posted on our Multithreaded Java programming discussion forum is some version of how to create a thread pool. In nearly every server application, the question of thread pools and work queues comes up. In this article, Brian Goetz explores the motivations for thread pools, some basic implementation and tuning techniques, and some common hazards to avoid.

[PDF] How to Design a Parallel Program, A sample thread pool (green boxes) with waiting tasks (blue) and completed tasks (yellow). In computer programming, a thread pool is a software design pattern for achieving concurrency of execution in a computer program. Often also called a replicated workers or worker-crew model, a thread pool maintains multiple One benefit of a thread pool over creating a new thread for   An object that represents each Fibonacci value is created, and the ThreadPoolCallback method is passed to QueueUserWorkItem, which assigns an available thread in the pool to execute the method. Because each Fibonacci object is given a semi-random value to compute, and because each thread will be competing for processor time, you cannot know in

Comments
  • Consider using Akka.
  • Sounds like a Striped Executor.
  • Are you required to process each group at the same time, or can one group wait for another to be processed?
  • @Dodd10x One group can wait for another group. While tasks within a group must be executed synchronously, tasks of different groups can be executed asynchronously. However, I worry about starvation: what if group A waits for group B, but there's always a new task for group B?
  • See the dispatch queue approach described in that thread. It allows groups to be serial, but the executor to process any work available.
  • how we gonna initialize future task, to begin with?
  • which exactly Akka library/class I have to look at?
  • I don't knnow, sorry. I haven't actually checked it, it just seemed promising. I ended up going with the Striped Executor Service.
  • But then I would have a singleThreadExecutor for each of my group of serial tasks, right? I've done that (not exactly that, but something like that), and it works great. The thing is: I can't have a thread for each of my groups, there are way too many groups. That's why I need a single thread pool, with N smart threads that understand my needs for serial execution for some particular tasks.
  • Well, thats a bit different, if you have too many groups then you probably need to a create a list of runnables and submit them to an n number of single threaded thread pools