CompletionService/ExecutorCompletionService/线程池/concurrent包

线程池

线程池的基本思想：线程频繁的创建、销毁会极大地占用系统资源，为了减少系统在创建销毁线程时的开销，线程池应运而生。线程池包括多个已创建的线程，当有任务要在新线程中执行时，将任务提交给线程池，线程池选取空闲线程或新开线程执行该任务，可见线程池应维护一个任务队列和线程队列。此外还要对线程最大数、最小数目、空闲等待时间等进行管理。

Executor、ExecutorService接口（线程池）

Executor提供一种将任务提交与每个任务将如何运行的机制（包括线程使用的细节、调度等）分离开来的方法。

方法execute()向线程池提交任务

ExecutorService接口 一个ExecutorService包括运行、关闭、终止三个生命阶段，当一个ExecutorService处于关闭状态时，不能再向线程池提交任务。线程池管理，提交任务功能扩展submit

AbstractExecutorService实现了ExecutorService接口 ，ThreadPoolExecutor、ScheduledThreadPoolExecutor类是具体的线程池实现类。

CompletionService接口、ExecutorCompletionService类

能返回已完成任务，维护一个已完成任务队列（Future），通过take() 函数可以获得任务。

CompletionService接口将生产新的异步任务与使用已完成任务的结果分离开来的服务。生产者 submit 执行的任务。使用者 take 已完成的任务，并按照完成这些任务的顺序处理它们的结果。

ExecutorCompletionService类 依赖于一个线程池对象Excutor，在一个Excutor基础上维护一个已完成任务队列。

Callable/Runnable、Future（任务接口），

（1）Callable是能够返回结果的可执行任务，Runnable不返回执行结果。文档中说Future 表示异步计算的结果，从CompletionService接口来理解，感觉不如说Future是执行完之后，包含返回结果且可以取消的任务。

（2）execute方法只能提交Runnable任务，ExecutorService接口扩展的submit方法还可以提交Callable任务，ExecutorCompletionService类的submit也支持两种任务

（3）FutrueTask实现了RunnableFuture接口，且可以由一个Callable对象来构造，基本包括了三个接口的功能

Executors（创建线程池）

类提供了用于此包中所提供的执行程序服务的工厂方法。

引用这里

ExecutorCompletionService详解

ExecutorCompletionService源码

//成员变量，一个执行器，一个存放执行结果的阻塞队列
//一个ExecutorService抽象类（转化submit的Runnable或Callable为RunnableFuture<V>）
private final Executor executor;
private final AbstractExecutorService aes;
private final BlockingQueue<Future<V>> completionQueue;


//提交待执行的任务，任务均会转化为RunnableFuture<V>接口，然后放入执行器执行。
public Future<V> submit(Callable<V> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<V> f = newTaskFor(task);
executor.execute(new QueueingFuture(f));
return f;
} public Future<V> submit(Runnable task, V result) {
if (task == null) throw new NullPointerException();
RunnableFuture<V> f = newTaskFor(task, result);
executor.execute(new QueueingFuture(f));
return f;
}


//内部类QueueingFuture，将转化后的RunnableFuture<V>接口对象，装配，重写done方法，使得任务执行结束后，放入完成队列
private class QueueingFuture extends FutureTask<Void> {
QueueingFuture(RunnableFuture<V> task) {
super(task, null);
this.task = task;
}
protected void done() { completionQueue.add(task); }
private final Future<V> task;
}


//阻塞方法，直至队列中有任务完成
public Future<V> take() throws InterruptedException {
return completionQueue.take();
} public Future<V> poll() {
return completionQueue.poll();
} public Future<V> poll(long timeout, TimeUnit unit)
throws InterruptedException {
return completionQueue.poll(timeout, unit);
}


阻塞队列take方法源码

//用到了ReentrantLock，队列为空时await，队列使用put加元素时signal.....好一个PV操作...
public E take() throws InterruptedException {
E x;
int c = -1;
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();
try {
while (count.get() == 0) {
notEmpty.await();
}
x = dequeue();
c = count.getAndDecrement();
if (c > 1)
notEmpty.signal();
} finally {
takeLock.unlock();
}
if (c == capacity)
signalNotFull();
return x;
}


总结

ExecutorCompletionService初始化时放入了一个线程池，并且每个submit的任务的执行结果，都会记录到本身的阻塞队列中，重点是，

如果使用take()阻塞方法取出结果的数量等于submit的任务数量，那么所有任务一定是complete（完成）。

代码示例

CompletionService<Integer> completionExecutor=new ExecutorCompletionService<>(Executors.newFixedThreadPool());List<Task> subTasks=multiTask.getSubTasks();//假设list中有5个任务//全部submit到ExecutorCompletionService
for (Task task : subTasks) {
completionExecutor.submit( new Runnable() {
public void run() {
task.runTask();
}
}, Integer.valueOf(TASKFINISHED));
}//如果取任务的数量恰好是submit的数量，那么下面这段一定阻塞，直至submit的任务全部完成
for (int i = 0; i < subTasks.size(); i++) {
try {
//阻塞，直至有一个任务完成，并取回结果
Future<Integer> future=completionExecutor.take();
future.get()
} catch (InterruptedException) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}