Java并发编程类学习三（线程的执行）

Executor

任务运行的执行器。可以解耦任务提交和计算的过程。换句话说，只管提交给执行器执行，而不用关系执行的内部过程。

任务被提交后是立即执行的。

public interface Executor {
void execute(Runnable command);
}

ExecutorService

ExecutorService继承了Executor接口。提供了管理任务终止的功能，并且能生成Future跟踪任务异步执行的结果。

调用shutdown方法后，将继续执行之前提交的任务直到完成，但是拒绝任务的提交。submit()方法扩展了execute方法，任务提交后提供Future返回，可以通过返回Future控制任务的生命周期。invokeAny和invokeAll适合大批量的提交。

public interface ExecutorService extends Executor

ThreadFactory

一个Thread生成的工厂接口，代替用构造器创建Thread，可以灵活的按自己需求创建Thread的子类。

public interface ThreadFactory {
Thread newThread(Runnable r);
}

DefaultThreadFactory

DefaultThreadFactory是ThreadFactory接口的默认实现。其中用到了原子变量类来生成线程名。

static class DefaultThreadFactory implements ThreadFactory {
private static final AtomicInteger poolNumber = new AtomicInteger(1);
private final ThreadGroup group;
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final String namePrefix;
DefaultThreadFactory() {
SecurityManager s = System.getSecurityManager();
group = (s != null) ? s.getThreadGroup() :
Thread.currentThread().getThreadGroup();
namePrefix = "pool-" +
poolNumber.getAndIncrement() +
"-thread-";
}
public Thread newThread(Runnable r) {
Thread t = new Thread(group, r,
namePrefix + threadNumber.getAndIncrement(),
0);
if (t.isDaemon())
t.setDaemon(false);
if (t.getPriority() != Thread.NORM_PRIORITY)
t.setPriority(Thread.NORM_PRIORITY);
return t;
}
}

ThreadPoolExecutor

ThreadPoolExecutor是一个线程池，扩展实现了ExecutorService的方法。它用线程池中可用的线程来处理提交的任务。正常情况下建议使用Executors。

ThreadPoolExecutor根据基本线程数和最大线程数自动调整当前线程数；当基本线程数和最大线程数相等时，就是一个固定数量的线程池。当线程数小于基本线程数，任务到达时，即使有空闲的线程，也会创建新的线程服务当前的任务。当最大线程数设置为无边界，如Integer.MAX_VALUE，线程池会随意调整当前并发任务的数量。

线程池初始化为0个线程，当任务到达时线程才被创建，但是你可以重写方法prestartCoreThread或者prestartAllCoreThreads。

默认用DefaultThreadFactory创建线程，可以传入自定义ThreadFactory。

队列机制：

当前线程数少于基本线程数时，执行器总是倾向于创建线程处理任务而不是将任务加入队列中等待处理；

当前线程大于或者等于基本线程数，执行器总是倾向于将任务加入队列中等待处理而不是创建线程处理任务；

如果任务不能加入队列，一个新的线程会被创建，前提是创建后的线程数不能大于最大线程数，否则这个任务会被拒绝处理；

public class ThreadPoolExecutor extends AbstractExecutorService {

Executors

一个功能强大的工具类，通过静态工厂方法来生成ExecutorService、ScheduledExecutorService、ThreadFactory和Callable的实例。

Executors在创建ExecutorService时调用了ThreadPoolExecutor的方法。

CompletionService

虽然ExecutorService可以在执行任务的时候，可以用Future来跟踪执行的结果。但是对于多个返回的执行结果，却需要去轮询哪个执行过程已经结束，这无疑浪费了资源。

CompletionService这个接口就是为了解决这个问题，take()方法会阻塞直到返回最先获取的执行结果Future。

public interface CompletionService<V> {
Future<V> submit(Callable<V> task);
Future<V> submit(Runnable task, V result);
Future<V> take() throws InterruptedException;
Future<V> poll();
Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException;
}

ExecutorCompletionService

ExecutorCompletionService是CompletionService接口的实现。内部采用了BlockingQueue来保存执行结果的Future，BlockingQueue是个阻塞队列，可以很好处理并发问题。

在构造ExecutorCompletionService的时候，必须传入Executor，内部任务的执行还是依赖Executor。

public class ExecutorCompletionService<V> implements CompletionService<V> {
private final Executor executor;
private final AbstractExecutorService aes;
private final BlockingQueue<Future<V>> completionQueue;
...

Exchanger

两个线程可以用Exchanger直接交换数据，而不用其他容器。

官方实例：

class FillAndEmpty {

Exchanger exchanger = new Exchanger();

DataBuffer initialEmptyBuffer = ... a made-up type

DataBuffer initialFullBuffer = ...

class FillingLoop implements Runnable {

public void run() {

DataBuffer currentBuffer = initialEmptyBuffer;

try {

while (currentBuffer != null) {

addToBuffer(currentBuffer);

if (currentBuffer.isFull())

currentBuffer = exchanger.exchange(currentBuffer);

}

} catch (InterruptedException ex) { ... handle ... }

}

}

class EmptyingLoop implements Runnable {

public void run() {

DataBuffer currentBuffer = initialFullBuffer;

try {

while (currentBuffer != null) {

takeFromBuffer(currentBuffer);

if (currentBuffer.isEmpty())

currentBuffer = exchanger.exchange(currentBuffer);

}

} catch (InterruptedException ex) { ... handle ...}

}

}

void start() {

new Thread(new FillingLoop()).start();

new Thread(new EmptyingLoop()).start();

}

}

}