定义一个线程类：
线程类需要实现Comparable接口，同时要定义需要比较的字段，
另外需要自己实现一个优先级的线程池如下：
package priority;
import com.google.common.collect.Lists;
import com.google.common.collect.Queues;
import com.google.common.util.concurrent.*;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.*;
import static com.google.common.base.Preconditions.checkArgument;
public class PriorityThreadPoolExecutorextends ThreadPoolExecutorimplements ListeningExecutorService {
public PriorityThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? long keepAliveTime, TimeUnit unit, BlockingQueue workQueue) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
? ? }
public PriorityThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? long keepAliveTime, TimeUnit unit, BlockingQueue workQueue,
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? RejectedExecutionHandler handler) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
? ? }
public PriorityThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? long keepAliveTime, TimeUnit unit, BlockingQueue workQueue,
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ThreadFactory threadFactory) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
? ? }
public PriorityThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? long keepAliveTime, TimeUnit unit, BlockingQueue workQueue,
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ThreadFactory threadFactory, RejectedExecutionHandler handler) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
? ? }
@Override
? ? protected RunnableFuturenewTaskFor(Runnable runnable, T value) {
return new ComparableFutureTask(runnable, value);
? ? }
@Override
? ? protected RunnableFuturenewTaskFor(Callable callable) {
return new ComparableFutureTask(callable);
? ? }
protected class ComparableFutureTask
extends FutureTaskimplements Comparable>,ListenableFuture? {
private Objectobject;
? ? ? ? // The execution list to hold our listeners.
? ? ? ? private final ExecutionListexecutionList =new ExecutionList();
? ? ? ? public ComparableFutureTask(Callable callable) {
super(callable);
? ? ? ? ? ? object = callable;
? ? ? ? }
public ComparableFutureTask(Runnable runnable, V result) {
super(runnable, result);
? ? ? ? ? ? object = runnable;
? ? ? ? }
@Override
? ? ? ? @SuppressWarnings("unchecked")
public int compareTo(ComparableFutureTask 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;
? ? ? ? }
@Override
? ? ? ? public void addListener(Runnable listener, Executor exec) {
executionList.add(listener, exec);
? ? ? ? }
/**
* Internal implementation detail used to invoke the listeners.
*/
? ? ? ? @Override
? ? ? ? protected void done() {
executionList.execute();
? ? ? ? }
}
@Override public ListenableFuturesubmit(Runnable task) {
ComparableFutureTask ftask =new ComparableFutureTask(task, null);
? ? ? ? execute(ftask);
? ? ? ? return ftask;
? ? }
@Override public ListenableFuturesubmit(Runnable task, T result) {
ComparableFutureTask ftask =new? ComparableFutureTask(task, result);
? ? ? ? execute(ftask);
? ? ? ? return ftask;
? ? }
@Override public ListenableFuturesubmit(Callable task) {
ComparableFutureTask ftask =new ComparableFutureTask(task);
? ? ? ? execute(ftask);
? ? ? ? return ftask;
? ? }
@Override public T invokeAny(Collection> tasks)
throws InterruptedException, ExecutionException {
try {
return invokeAnyImpl(this, tasks, false, 0);
? ? ? ? }catch (TimeoutException cannotHappen) {
throw new AssertionError();
? ? ? ? }
}
@Override public T invokeAny(
Collection> tasks, long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
return invokeAnyImpl(this, tasks, true, unit.toNanos(timeout));
? ? }
@Override public List>invokeAll(Collection> tasks)
throws InterruptedException {
if (tasks ==null) {
throw new NullPointerException();
? ? ? ? }
List> futures =new ArrayList>(tasks.size());
? ? ? ? boolean done =false;
? ? ? ? try {
for (Callable t : tasks) {
ComparableFutureTask f =new ComparableFutureTask(t);
? ? ? ? ? ? ? ? futures.add(f);
? ? ? ? ? ? ? ? execute(f);
? ? ? ? ? ? }
for (Future f : futures) {
if (!f.isDone()) {
try {
f.get();
? ? ? ? ? ? ? ? ? ? }catch (CancellationException ignore) {
}catch (ExecutionException ignore) {
}
}
}
done =true;
? ? ? ? ? ? return futures;
? ? ? ? }finally {
if (!done) {
for (Future f : futures) {
f.cancel(true);
? ? ? ? ? ? ? ? }
}
}
}
@Override public List>invokeAll(
Collection> tasks, long timeout, TimeUnit unit)
throws InterruptedException {
if (tasks ==null || unit ==null) {
throw new NullPointerException();
? ? ? ? }
long nanos = unit.toNanos(timeout);
? ? ? ? List> futures =new ArrayList>(tasks.size());
? ? ? ? boolean done =false;
? ? ? ? try {
for (Callable t : tasks) {
futures.add(new ComparableFutureTask(t));
? ? ? ? ? ? }
long lastTime = System.nanoTime();
? ? ? ? ? ? // Interleave time checks and calls to execute in case
// executor doesn't have any/much parallelism.
? ? ? ? ? ? Iterator> it = futures.iterator();
? ? ? ? ? ? while (it.hasNext()) {
execute((Runnable) (it.next()));
? ? ? ? ? ? ? ? long now = System.nanoTime();
? ? ? ? ? ? ? ? nanos -= now - lastTime;
? ? ? ? ? ? ? ? lastTime = now;
? ? ? ? ? ? ? ? if (nanos <=0) {
return futures;
? ? ? ? ? ? ? ? }
}
for (Future f : futures) {
if (!f.isDone()) {
if (nanos <=0) {
return futures;
? ? ? ? ? ? ? ? ? ? }
try {
f.get(nanos, TimeUnit.NANOSECONDS);
? ? ? ? ? ? ? ? ? ? }catch (CancellationException ignore) {
}catch (ExecutionException ignore) {
}catch (TimeoutException toe) {
return futures;
? ? ? ? ? ? ? ? ? ? }
long now = System.nanoTime();
? ? ? ? ? ? ? ? ? ? nanos -= now - lastTime;
? ? ? ? ? ? ? ? ? ? lastTime = now;
? ? ? ? ? ? ? ? }
}
done =true;
? ? ? ? ? ? return futures;
? ? ? ? }finally {
if (!done) {
for (Future f : futures) {
f.cancel(true);
? ? ? ? ? ? ? ? }
}
}
}
/*
* This following method is a modified version of one found in
* http://gee.cs.oswego.edu/cgi-bin/viewcvs.cgi/jsr166/src/test/tck/AbstractExecutorServiceTest.java?revision=1.30
* which contained the following notice:
*
* Written by Doug Lea with assistance from members of JCP JSR-166
* Expert Group and released to the public domain, as explained at
* http://creativecommons.org/publicdomain/zero/1.0/
* Other contributors include Andrew Wright, Jeffrey Hayes,
* Pat Fisher, Mike Judd.
*/
? ? /**
? ? * An implementation of {@link ExecutorService#invokeAny} for {@link ListeningExecutorService}
* implementations.
? ? */ static T invokeAnyImpl(ListeningExecutorService executorService,
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? Collection> tasks, boolean timed, long nanos)
throws InterruptedException, ExecutionException, TimeoutException {
int ntasks = tasks.size();
? ? ? ? checkArgument(ntasks >0);
? ? ? ? List> futures = Lists.newArrayListWithCapacity(ntasks);
? ? ? ? BlockingQueue> futureQueue = Queues.newLinkedBlockingQueue();
? ? ? ? // For efficiency, especially in executors with limited
// parallelism, check to see if previously submitted tasks are
// done before submitting more of them. This interleaving
// plus the exception mechanics account for messiness of main
// loop.
? ? ? ? try {
// Record exceptions so that if we fail to obtain any
// result, we can throw the last exception we got.
? ? ? ? ? ? ExecutionException ee =null;
? ? ? ? ? ? long lastTime = timed System.nanoTime() :0;
? ? ? ? ? ? Iterator> it = tasks.iterator();
? ? ? ? ? ? futures.add(submitAndAddQueueListener(executorService, it.next(), futureQueue));
? ? ? ? ? ? --ntasks;
? ? ? ? ? ? int active =1;
? ? ? ? ? ? for (;;) {
Future f = futureQueue.poll();
? ? ? ? ? ? ? ? if (f ==null) {
if (ntasks >0) {
--ntasks;
? ? ? ? ? ? ? ? ? ? ? ? futures.add(submitAndAddQueueListener(executorService, it.next(), futureQueue));
? ? ? ? ? ? ? ? ? ? ? ? ++active;
? ? ? ? ? ? ? ? ? ? }else if (active ==0) {
break;
? ? ? ? ? ? ? ? ? ? }else if (timed) {
f = futureQueue.poll(nanos, TimeUnit.NANOSECONDS);
? ? ? ? ? ? ? ? ? ? ? ? if (f ==null) {
throw new TimeoutException();
? ? ? ? ? ? ? ? ? ? ? ? }
long now = System.nanoTime();
? ? ? ? ? ? ? ? ? ? ? ? nanos -= now - lastTime;
? ? ? ? ? ? ? ? ? ? ? ? lastTime = now;
? ? ? ? ? ? ? ? ? ? }else {
f = futureQueue.take();
? ? ? ? ? ? ? ? ? ? }
}
if (f !=null) {
--active;
? ? ? ? ? ? ? ? ? ? try {
return f.get();
? ? ? ? ? ? ? ? ? ? }catch (ExecutionException eex) {
ee = eex;
? ? ? ? ? ? ? ? ? ? }catch (RuntimeException rex) {
ee =new ExecutionException(rex);
? ? ? ? ? ? ? ? ? ? }
}
}
if (ee ==null) {
ee =new ExecutionException(null);
? ? ? ? ? ? }
throw ee;
? ? ? ? }finally {
for (Future f : futures) {
f.cancel(true);
? ? ? ? ? ? }
}
}
/**
? ? * Submits the task and adds a listener that adds the future to {@code queue} when it completes.
*/
? ? private static ListenableFuturesubmitAndAddQueueListener(
ListeningExecutorService executorService, Callable task,
? ? ? ? ? ? final BlockingQueue> queue) {
final ListenableFuture future = executorService.submit(task);
? ? ? ? future.addListener(new Runnable() {
@Override public void run() {
queue.add(future);
? ? ? ? ? ? }
}, MoreExecutors.sameThreadExecutor());
? ? ? ? return future;
? ? }
}
测试方法如下：
public static void main(String[] args) {
PriorityBlockingQueue priorityQueue =new PriorityBlockingQueue(1000);
? ? ThreadPoolExecutor threadPoolExecutor =new PriorityThreadPoolExecutor(2, 2, 1000, TimeUnit.MILLISECONDS,priorityQueue);
? ? ListeningExecutorService processExecutorService = MoreExecutors.listeningDecorator(threadPoolExecutor);
? ? for (int i =0; i <15; i++) {
Random random =new Random();
? ? ? ? ProcessEntry threadA =new ProcessEntry(random.nextInt(30));
? ? ? ? ListenableFuture submit = processExecutorService.submit(threadA);
? ? }
System.out.println("完成");
? ? try {
Thread.sleep(5000);
? ? }catch (InterruptedException e) {
e.printStackTrace();
? ? }
System.out.println("插入小的数字");
? ? ProcessEntry threadb =new ProcessEntry(-1);
? ? processExecutorService.submit(threadb);
? ? System.out.println("插入小的数字2");
? ? ProcessEntry threadb1 =new ProcessEntry(-30);
? ? processExecutorService.submit(threadb1);
? ? System.out.println("插入 大的数字");
? ? ProcessEntry threadA =new ProcessEntry(200);
? ? ListenableFuture processFuture = processExecutorService.submit(threadA);
? ? Futures.addCallback(processFuture, new FutureCallback() {
@Override
? ? ? ? public void onSuccess(Object o) {
System.out.println("200 成功");
? ? ? ? }
@Override
? ? ? ? public void onFailure(Throwable throwable) {
throwable.printStackTrace();
? ? ? ? ? ? System.out.println("200 失败");
? ? ? ? }
});
}
结果：
完成
priority:29
priority:6
插入小的数字
插入小的数字2
插入 大的数字
priority:23
priority:28
priority:23
priority:200
200 成功
priority:22
priority:22
priority:17
priority:16
priority:16
priority:15
priority:7
priority:3
priority:3
priority:1
priority:-1
priority:-30
可以看到是按照从大到小的顺序执行的线程