How do I implement task prioritization using an ExecutorService in Java 5?

by

I have solved this problem in a reasonable fashion, and I’ll describe it below for future reference to myself and anyone else who runs into this problem with the Java Concurrent libraries.

Using a PriorityBlockingQueue as the means for holding onto tasks for later execution is indeed a movement in the correct direction. The problem is that the PriorityBlockingQueue must be generically instantiated to contain Runnable instances, and it is impossible to call compareTo (or similiar) on a Runnable interface.

Onto solving the problem. When creating the Executor, it must be given a PriorityBlockingQueue. The queue should further be given a custom Comparator to do proper in place sorting:

new PriorityBlockingQueue<Runnable>(size, new CustomTaskComparator());

Now, a peek at CustomTaskComparator:

public class CustomTaskComparator implements Comparator<MyType> {

    @Override
    public int compare(MyType first, MyType second) {
         return comparison;
    }

}

Everything looking pretty straight forward up to this point. It gets a bit sticky here. Our next problem is to deal with the creation of FutureTasks from the Executor. In the Executor, we must override newTaskFor as so:

@Override
protected <V> RunnableFuture<V> newTaskFor(Callable<V> c) {
    //Override the default FutureTask creation and retrofit it with
    //a custom task. This is done so that prioritization can be accomplished.
    return new CustomFutureTask(c);
}

Where c is the Callable task that we’re trying to execute. Now, let’s have a peek at CustomFutureTask:

public class CustomFutureTask extends FutureTask {

    private CustomTask task;

    public CustomFutureTask(Callable callable) {
        super(callable);
        this.task = (CustomTask) callable;
    }

    public CustomTask getTask() {
        return task;
    }

}

Notice the getTask method. We’re gonna use that later to grab the original task out of this CustomFutureTask that we’ve created.

And finally, let’s modify the original task that we were trying to execute:

public class CustomTask implements Callable<MyType>, Comparable<CustomTask> {

    private final MyType myType;

    public CustomTask(MyType myType) {
        this.myType = myType;
    }

    @Override
    public MyType call() {
        //Do some things, return something for FutureTask implementation of `call`.
        return myType;
    }

    @Override
    public int compareTo(MyType task2) {
        return new CustomTaskComparator().compare(this.myType, task2.myType);
    }

}

You can see that we implement Comparable in the task to delegate to the actual Comparator for MyType.

And there you have it, customized prioritization for an Executor using the Java libraries! It takes some bit of bending, but it’s the cleanest that I’ve been able to come up with. I hope this is helpful to someone!

More Related Contents:

  1. How to wait for all threads to finish, using ExecutorService?
  2. Is there an advantage to use a Synchronized Method instead of a Synchronized Block?
  3. Why is creating a Thread said to be expensive?
  4. When and how should I use a ThreadLocal variable?
  5. What is the difference between Thread.start() and Thread.run()?
  6. Java Wait and Notify: IllegalMonitorStateException
  7. Are static variables shared between threads?
  8. How do synchronized static methods work in Java and can I use it for loading Hibernate entities?
  9. Is iterating ConcurrentHashMap values thread safe?
  10. Why use a ReentrantLock if one can use synchronized(this)?
  11. Is it safe to get values from a java.util.HashMap from multiple threads (no modification)?
  12. What is mutex and semaphore in Java ? What is the main difference?
  13. CountDownLatch vs. Semaphore
  14. Constructor synchronization in Java
  15. Parallel execution of directed acyclic graph of tasks
  16. Javafx: Difference between javafx.concurent and Platform.runLater?
  17. Instruction reordering & happens-before relationship [duplicate]
  18. Programmatic deadlock detection in java
  19. When should we use Java’s Thread over Executor?
  20. Why does java.util.concurrent.ArrayBlockingQueue use ‘while’ loops instead of ‘if’ around calls to await()?
  21. Run Java Threads sequentially
  22. WAITING at sun.misc.Unsafe.park(Native Method)
  23. ThreadPoolExecutor Block When its Queue Is Full?
  24. Java : Does wait() release lock from synchronized block
  25. How Threadpool re-use Threads and how it works
  26. Concurrent JUnit testing
  27. Handling Exceptions for ThreadPoolExecutor
  28. notifyAll() throws IllegalMonitorStateException
  29. Will a chain of method calls (CompletableFuture API) execute asynchronously if the first method in a chain is asynchronous?
  30. ExecutorCompletionService? Why do need one if we have invokeAll?

Leave a Comment