java并发学习之Executor源码解析
作者：pq217
Runnable && Thread
Runnable和Thread都是java.lang包最基本的线程操作类，相当于官方的，而Executor接口及其实现都是Doug Lea写的java.util.concurrent包下，属于民间的，当然因为太牛逼了所以也在jdk中
先看官方提供的线程操作，其中Runnable是一个函数式接口
@FunctionalInterface
public interface Runnable {
public abstract void run();
}
可以理解为一个待执行的函数，或者理解为一个任务(通过调用run方法可以实际的执行任务)
Runable是一个定义的任务，而Thread是它的一个执行者，它提供start方法可以开启一个新线程执行传入的Runable任务，这很像命令模式，用户通过实现Runable制定一个命令，交给Thread这个执行者去具体执行
Runnable && Thread
所以一般开启新线程执行方法的方式如下
Runnable task = () -> {
// do something
};
new Thread(task).start(); // 开启新线程执行而开启新线程执行方法也只有这一个途径可走，就是必须通过官方的Thread.start方法
Executor
虽然开启线程执行任务只能走Thread.start方法，方法只有一个，但我们能做的是可以改变任务运行的方式，比如我们可以决定什么时候执行任务，多少个任务共用某个线程排队工作
最具代表性的就是线程池，线程池只是修改了任务执行的方式：即所有任务共用固定数量的线程，但最终的运行终归还是通过Thread.start方法实际在线程中执行Runnable方法
Doug Lea所写的java.util.concurrent.Executor把各种方式的Runnable执行器的一个抽象
public interface Executor {
void execute(Runnable command);
}
Executor
而且提供了一些常用的执行器供我们使用，比如ThreadPoolExecutor(线程池)，ForkJoinPool，当然我们也可以自己定义一个Executor按照自己的方式执行任务，比如netty中实现的SingleThreadEventExecutor是一种单个线程依次处理所有任务的执行器
ExecutorService
如果说Executor是一种对按自己套路执行任务的执行器，是一种抽象分类，那么ExecutorService就是其下的一个子分类，它是一种特殊的执行器，从名字直译来看："执行器服务"，从一个执行器升级为执行服务，像不像某公司从卖产品业务升级为产品安装售后整套服务
所以ExecutorService作为一个特殊的服务类Executor，不光能按照自己的方式执行任务，还推出了一系列附加"服务"，那就看看这种特殊的执行器都提供了什么服务
public interface ExecutorService extends Executor {
void shutdown();
Future<?> submit(Runnable task);
<T> Future<T> submit(Callable<T> task);
......
}只贴了些重要方法，首先继承了Executor肯定是要继承void execute(Runnable) 方法代表它首先是一个任务执行器
shutdown方法代表这个执行器是有状态的，可以关闭服务的
然后就是重量级的submit方法，这也是ExecutorService提供的最具特色的服务，如果打开ExecutorService的类，注释第一句就写着：
/**
* An {@link Executor} that provides methods to manage termination and
* methods that can produce a {@link Future} for tracking progress of
* one or more asynchronous tasks.翻译过来大致就是ExecutorService是一个特殊的Executor执行器，他可以终止服务并且可以创建一个Future来跟踪任务执行进度
ExecutorService的submit不光能接受Runnable，还可以接受一种新型任务形式：Callable，即有返回结果和异常的任务
Callable
上面我们总结Runnable是一种可执行任务，而这种任务是没有返回结果的，也不能抛出异常，很显然现实中很多任务是需要有返回结果的，比如计算1+1等于几的任务，所以为了扩展任务类型，Doug Lea又定义一种新的任务：Callable，而ExecutorService可以接受并处理这样的任务
@FunctionalInterface
public interface Callable<V> {
V call() throws Exception;
}Future
submit执行的返回值是一个Future，从注释看出他可以跟踪任务的进度，它就好比任务的一个订单，通过订单可以取消任务，查看任务进度等
- boolean cancel(boolean) 取消
取消任务，就好比在某宝买了个东西，本质就是提交一个"把东西给我送过来"的任务，而通过订单我们就可以取消这个任务 - isCancelled() && isDone()
查看任务状态，是否取消和是否完成 - V get()
获取任务执行结果，如果没完成则阻塞，如果是Runnable，返回的就是null
ExecutorService
AbstractExecutorService
ExecutorService制订了一种新型执行器，它的特殊在于可以跟踪任务进度甚至取消任务，那么如何实现呐
首先execute方法只会单纯的执行任务，按照自己执行器的逻辑调用Thread.start方法，不会有返回值，也不支持跟踪进度或取消
所以解决方案只有一个：调包任务，具体这样操作：当用户提交任务，不是直接去execute执行，而是把任务包装为一个新任务，新任务执行原任务的同时，还负责获取任务结果，跟踪任务状态等工作，相当于是原任务的一个代理
以上即是AbstractExecutorService负责的工作，它是ExecutorService关于任务跟踪业务的相关实现(解决方案即是代理任务)，继承了AbstractExecutorService的任务执行器既可以实现跟踪任务的功能
AbstractExecutorService
来看一下AbstractExecutorService的submit方法
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
// 生成一个依赖于原任务的新任务
RunnableFuture<Void> ftask = newTaskFor(task, null);
// 执行新任务
execute(ftask);
// 返回新任务(充当任务跟踪器)
return ftask;
}其中newTaskFor方法负责生成新任务，同时也是原任务的跟踪器(订单)
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
}所以整个执行过程就是把Runnable或Callable转换为FutureTask的过程，而FutureTask首先是一个新任务(继承Runnable)，又是原任务的跟踪器(继承Future)，这种任务归类为RunnableFuture
public interface RunnableFuture<V> extends Runnable, Future<V> {
void run();
}FutureTask
接下来就看FutureTask是如何实现跟踪替换原任务的，首先它的重点属性如下
- int state; 存储原任务的执行状态
- Callable<V> callable; 原任务Runnable也可以适配为返回null的Callable
- Object outcome; 原任务的返回结果
再看一下重点方法
1.初始化，存储原任务，任务状态为NEW
public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW; // ensure visibility of callable
}2.run()，真实的被执行任务，加入状态判断，执行原任务通过try-catch获取异常，通过outcome保存结果
public void run() {
// 如果任务不是新状态，直接返回
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
// 获取原任务
Callable<V> c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
try {
// 执行原任务
result = c.call();
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
// 有异常设置异常
setException(ex);
}
if (ran)
// 设置outcome存储返回值
set(result);
}
} finally {
// runner must be non-null until state is settled to
// prevent concurrent calls to run()
runner = null;
// state must be re-read after nulling runner to prevent
// leaked interrupts
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}3.get()，获取结果，如果状态未完成阻塞等待，完成则返回outcome
public V get() throws InterruptedException, ExecutionException {
int s = state;
// 如果任务未完成，阻塞等待
if (s <= COMPLETING)
s = awaitDone(false, 0L);
// 任务完成，返回结果
return report(s);
}
private V report(int s) throws ExecutionException {
Object x = outcome;
if (s == NORMAL)
return (V)x; // 返回结果即outcome
if (s >= CANCELLED)
throw new CancellationException();
throw new ExecutionException((Throwable)x);
}4.cancel()，只要把状态设置为CANCEL，run时就会直接return而不会执行原任务
public boolean cancel(boolean mayInterruptIfRunning) {
// 状态变为CANCELLED
if (!(state == NEW &&
UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
return false;
// 如果正在运行调用interrupt阻断正在执行的任务
try { // in case call to interrupt throws exception
if (mayInterruptIfRunning) {
try {
Thread t = runner;
if (t != null)
t.interrupt();
} finally { // final state
UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
}
}
} finally {
finishCompletion();
}
return true;
}总结
AbstractExecutorService只是实现了让任务变得可跟踪，通过调包任务，而具体任务的执行最终依然会调用execute，这个方法AbstractExecutorService并没有实现，因为这也是所有执行器的差异所在，即按自己的方式选择线程执行任务， 比如ThreadPoolExecutor线程池的固定线程数执行所有任务的模式，也就是只需要实现execute方法，而submit则交给父类AbstractExecutorService处理
以上就是java并发学习之Executor源码解析的详细内容，更多关于java并发Executor的资料请关注脚本之家其它相关文章！