在ThreadPoolExecutor中测试PriorityBlockingQueue

Question

我在这个示例中实现了具有PriorityBlockingQueue的ThreadPoolExecutor：http://codingdict.com/questions/131378

并写了一个测试：

PriorityExecutor executorService = (PriorityExecutor)  PriorityExecutor.newFixedThreadPool(16);
    executorService.submit(new Runnable() {
        @Override
        public void run() {
            try {
                Thread.sleep(1000);
                Thread.sleep(1000);
                System.out.println("1");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }, 1);

    executorService.submit(new Runnable() {
        @Override
        public void run() {
            try {
                Thread.sleep(1000);
                Thread.sleep(1000);
                System.out.println("3");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }, 3);

    executorService.submit(new Runnable() {
        @Override
        public void run() {
            try {
                Thread.sleep(1000);
                Thread.sleep(1000);
                System.out.println("2");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }, 2);

    executorService.submit(new Runnable() {
        @Override
        public void run() {
            try {
                Thread.sleep(1000);
                Thread.sleep(1000);
                System.out.println("5");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }, 5);

    executorService.submit(new Runnable() {
        @Override
        public void run() {
            try {
                Thread.sleep(1000);
                Thread.sleep(1000);
                System.out.println("4");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }, 4);

    executorService.shutdown();
    try {
        executorService.awaitTermination(30, TimeUnit.MINUTES);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }

但是最后，我没有得到1 2 3 4 5，我得到了这些数字的随机顺序。测试是否有问题，还是其他？如果首先，如何正确测试？

Answer 1

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

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

请参见下面的工作示例，以了解要实现的目标（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);
        }
    }
}