ThreadPoolExecutor的PriorityBlockingQueue支持問題

最近在使用ThreadPoolExecutor時遇到一個問題：當ThreadPoolExecutor使用的BlockingQueue爲PriorityBlockingQueue時，會出現異常，緣由是java.util.concurrent.FutureTask cannot be cast to java.lang.Comparable。Google之，發現有不少一樣的問題，但沒有給出解決方案，只能查看源代碼以期能找到並解決問題。

首先根據Exception找到問題緣由：

Exception in thread "main" java.lang.ClassCastException: java.util.concurrent.FutureTask cannot be cast to java.lang.Comparable
	at java.util.concurrent.PriorityBlockingQueue.siftUpComparable(PriorityBlockingQueue.java:347)
	at java.util.concurrent.PriorityBlockingQueue.offer(PriorityBlockingQueue.java:475)
	at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1329)
	at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:132)
...

找到java.util.concurrent.PriorityBlockingQueue.siftUpComparable方法：

private static <T> void siftUpComparable(int k, T x, Object[] array) {
        Comparable<? super T> key = (Comparable<? super T>) x;
        while (k > 0) {
            int parent = (k - 1) >>> 1;
            Object e = array[parent];
            if (key.compareTo((T) e) >= 0)
                break;
            array[k] = e;
            k = parent;
        }
        array[k] = key;
    }

是在Comparable<? super T> key = (Comparable<? super T>) x;上出現問題，根據java.util.concurrent.FutureTask cannot be cast to java.lang.Comparable知道x的類型是java.util.concurrent.FutureTask。如今看看FutureTask：

public class FutureTask<V> implements RunnableFuture<V> {
...

public interface RunnableFuture<V> extends Runnable, Future<V> {
...

public
interface Runnable {
...

public interface Future<V> {
...

可見FutureTask的確沒有實現Comparable接口。可是我所提交的Task

public static class Task implements Callable<Integer>, Comparable<Task> {
...

是實現了Comparable接口的。個人Task爲何變成了FutureTask？只好找到ThreadPoolExecutor的submit(Callable<T> task)方法一看究竟，它是在AbstractExecutorService中實現的。

public <T> Future<T> submit(Callable<T> task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task);
        execute(ftask);
        return ftask;
    }

能夠看到經過newTaskFor方法，我所提交的Task變成了FutureTask：

protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        return new FutureTask<T>(callable);
    }

OK，看來問題就是出在FutureTask這個類上了：

public class FutureTask<V> implements RunnableFuture<V> {
    private final Sync sync;

    public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        sync = new Sync(callable);
    }
...
    private final class Sync extends AbstractQueuedSynchronizer {
        private final Callable<V> callable;

        Sync(Callable<V> callable) {
            this.callable = callable;
        }
...

看來他是將我提交的Task的Comparable接口直接忽略了。如今的問題就變成了讓FutureTask支持Comparable接口，最簡單的方法是用一個ComparableFutureTask繼承FutureTask並實現Comparable接口，但也必需要Override ThreadPoolExecutor的newTaskFor方法，顯得有些麻煩。爲此又繼續Google，卻沒有發現一些更好的方法。先實現了再說。

package canghailan.util.concurrent;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author canghailan
 * @datetime 2011-12-10 13:57:19
 */
public class XThreadPoolExecutor extends ThreadPoolExecutor {

    public XThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
            long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    public XThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
            long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue,
            RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
    }

    public XThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
            long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue,
            ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
    }

    public XThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
            long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue,
            ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
    }

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

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

    protected class ComparableFutureTask<V>
            extends FutureTask<V> implements Comparable<ComparableFutureTask<V>> {
        private Object object;

        public ComparableFutureTask(Callable<V> callable) {
            super(callable);
            object = callable;
        }

        public ComparableFutureTask(Runnable runnable, V result) {
            super(runnable, result);
            object = runnable;
        }

        @Override
        @SuppressWarnings("unchecked")
        public int compareTo(ComparableFutureTask<V> o) {
            if (this == o) {
                return 0;
            }
            if (o == null) {
                return -1; // high priority
            }
            if (object != null && o.object != null) {
                if (object.getClass().equals(o.object.getClass())) {
                    if (object instanceof Comparable) {
                        return ((Comparable) object).compareTo(o.object);
                    }
                }
            }
            return 0;
        }
    }
}

測試一下：

package canghailan;

import canghailan.util.concurrent.XThreadPoolExecutor;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Phaser;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @author canghailan
 * @datetime 2011-12-10 7:09:39
 */
public class TestXThreadPoolExecutor {

    public static void main(String args[]) throws Exception {
        Phaser phaser = new Phaser(1);
        ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<>();
        ExecutorService service = new XThreadPoolExecutor(
                5, 5, 0L, TimeUnit.MILLISECONDS, new PriorityBlockingQueue<Runnable>());
        for (int i = 0; i < 10; ++i) {
            phaser.register();
            service.submit(new Task(i, queue, phaser));
        }
        phaser.arriveAndAwaitAdvance();
        System.out.println(queue);
        service.shutdown();
    }

    public static class Task implements Callable<Integer>, Comparable<Task> {
        private Integer id;
        private ConcurrentLinkedQueue<Integer> queue;
        private Phaser phaser;

        public Task(Integer id, ConcurrentLinkedQueue<Integer> queue, Phaser phaser) {
            this.id = id;
            this.queue = queue;
            this.phaser = phaser;
        }

        @Override
        public Integer call() throws Exception {
            queue.offer(id);
            phaser.arrive();
            return id;
        }

        @Override
        public int compareTo(Task o) {
            return -id.compareTo(o.id);
        }
    }
}

輸出爲[0, 1, 2, 3, 9, 8, 7, 6, 5, 4]，[0, 1, 2, 3, 4, 7, 6, 5, 9, 8]，[0, 1, 2, 3, 4, 7, 6, 5, 9, 8]...。

根據結果來看，的確起做用了。先到此爲止，若是有更好的辦法，請不吝賜教。