在ThreadPoolExecutor中测试PriorityBlockingQueue
问题内容:
我在这个示例中实现了具有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,我得到了这些数字的随机顺序。测试是否有问题,还是其他?如果首先,如何正确测试?
问题答案:
仅当池完全繁忙并且您提交了几个新任务时才考虑优先级。如果仅用一个线程定义池,则应获得预期的输出。在您的示例中,所有任务同时执行,并且首先完成的任务有些随机。
顺便说一下,如果您的队列已满并且您提交了新任务,则链接的实现会出现问题并引发异常。
请参见下面的工作示例,以了解要实现的目标(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);
}
}
}