在ThreadPoolExecutor中测试PriorityBlockingQueue

在ThreadPoolExecutor中测试PriorityBlockingQueue

仅当池完全繁忙并且您提交了几个新任务时才考虑优先级。如果仅用一个线程定义池，则应获得预期的输出。在您的示例中，所有任务同时执行，并且首先完成的任务有些随机。

顺便说一下，如果您的队列已满并且您提交了新任务，则链接的实现会出现问题并引发异常。

请参见下面的工作示例，以了解要实现的目标（

newTaskFor

为了使之正常工作，我已经以一种简单化的方式进行了覆盖-您可能需要改进该部分）。

打印：

1 2 3 4 5

。

public class Test {    public static void main(String[] args) {        PriorityExecutor executorService = (PriorityExecutor) PriorityExecutor.newFixedThreadPool(1);        executorService.submit(getRunnable("1"), 1);        executorService.submit(getRunnable("3"), 3);        executorService.submit(getRunnable("2"), 2);        executorService.submit(getRunnable("5"), 5);        executorService.submit(getRunnable("4"), 4);        executorService.shutdown();        try { executorService.awaitTermination(30, TimeUnit.MINUTES);        } catch (InterruptedException e) { e.printStackTrace();        }    }    public static Runnable getRunnable(final String id) {        return new Runnable() { @Override public void run() {     try {         Thread.sleep(1000);         System.out.println(id);     } catch (InterruptedException e) {         e.printStackTrace();     } }        };    }    static class PriorityExecutor extends ThreadPoolExecutor {        public PriorityExecutor(int corePoolSize, int maximumPoolSize,          long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);        }        //Utitlity method to create thread pool easily        public static ExecutorService newFixedThreadPool(int nThreads) { return new PriorityExecutor(nThreads, nThreads, 0L,       TimeUnit.MILLISECONDS, new PriorityBlockingQueue<Runnable>());        }        //Submit with New comparable task        public Future<?> submit(Runnable task, int priority) { return super.submit(new ComparableFutureTask(task, null, priority));        }        //execute with New comparable task        public void execute(Runnable command, int priority) { super.execute(new ComparableFutureTask(command, null, priority));        }        @Override        protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) { return (RunnableFuture<T>) callable;        }        @Override        protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) { return (RunnableFuture<T>) runnable;        }    }    static class ComparableFutureTask<T> extends FutureTask<T> implements Comparable<ComparableFutureTask<T>> {        volatile int priority = 0;        public ComparableFutureTask(Runnable runnable, T result, int priority) { super(runnable, result); this.priority = priority;        }        public ComparableFutureTask(Callable<T> callable, int priority) { super(callable); this.priority = priority;        }        @Override        public int compareTo(ComparableFutureTask<T> o) { return Integer.valueOf(priority).compareTo(o.priority);        }    }}