How to implement PriorityBlockingQueue with ThreadPoolExecutor and custom tasks

user_367101 0
Viewed 22366

I've searched a lot but could not find a solutuion to my problem.

I have my own class, BaseTask, that uses a ThreadPoolExecutor to handle tasks. I want task prioritization, but when I try to use a PriorityBlockingQueue I get ClassCastException because the ThreadPoolExecutor wraps my Tasks into a FutureTask object.

This obviously makes sense because the FutureTask does not implement Comparable, but how would I go on to solve the priority problem? I've read that you could override newTaskFor() in ThreadPoolExecutor, but I can not seem to find this method at all...?

Any suggestions would be much appreciated!

Some code to help:

In my BaseTask class I have

private static final BlockingQueue<Runnable> sWorkQueue = new PriorityBlockingQueue<Runnable>();

private static final ThreadFactory sThreadFactory = new ThreadFactory() {
    private final AtomicInteger mCount = new AtomicInteger(1);

    public Thread newThread(Runnable r) {
        return new Thread(r, "AsyncTask #" + mCount.getAndIncrement());
    }
};

private static final BaseThreadPoolExecutor sExecutor = new BaseThreadPoolExecutor(
    1, Integer.MAX_VALUE, 10, TimeUnit.SECONDS, sWorkQueue, sThreadFactory);

private final BaseFutureTask<Result> mFuture;

public BaseTask(int priority) {
    mFuture = new BaseFutureTask<Result>(mWorker, priority);
}

public final BaseTask<Params, Progress, Result> execute(Params... params) {

    /* Some unimportant code here */

    sExecutor.execute(mFuture);
}

In BaseFutureTask class

@Override
public int compareTo(BaseFutureTask another) {
    long diff = this.priority - another.priority;

    return Long.signum(diff);
}

In BaseThreadPoolExecutor class i override the 3 submit methods... The constructor in this class gets called, but none of the submit methods

futuretask executor priority-queue threadpool java
7 Answers

This answer is a simplified version of @StanislavVitvitskyy's answer. Thanks to him.

I wanted to make the jobs that I submitted be Comparable. I created an ExecutorService with a PriorityBlockingQueue and extend it to handle the newTaskFor(...) methods:

ExecutorService pool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize,
    keepAliveTime, timeUnit, new PriorityBlockingQueue<Runnable>()) {

    @Override
    protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
        return new ComparableFutureTask<T>(runnable, value);
    }

    @Override
    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        return new ComparableFutureTask<T>(callable);
    };
};

I defined a ComparableFutureTask which extends FutureTask and implements Comparable by delegating to the job.compareTo(...) that are submitted to the pool.

public class ComparableFutureTask<T> extends FutureTask<T>
    implements Comparable<Object> {

    private final Comparable<Object> comparableJob;

    @SuppressWarnings("unchecked")
    public ComparableFutureTask(Runnable runnable, T value) {
        super(runnable, value);
        this.comparableJob = (Comparable<Object>) runnable;
    }

    @SuppressWarnings("unchecked")
    public ComparableFutureTask(Callable<T> callable) {
        super(callable);
        this.comparableJob = (Comparable<Object>) callable;
    }

    @Override
    public int compareTo(Object o) {
        return this.comparableJob
            .compareTo(((ComparableFutureTask<?>) o).comparable);
    }
}

This ExecutorService then can handle Runnable or Callable jobs that are also Comparable. For example:

public class MyJob implements Runnable, Comparable<MyJob> {
    private int priority;
    ...
    @Override
    public int compareTo(MyJob other) {
        // we want higher priority to go first
        return other.priority - this.priority;
    }
    ...
}

It is important to note that if you submit a job that is not Comparable to this queue, it will throw a ClassCastException.

user_1798500
Related