并发编程(十二)—— Java 线程池 实现原理与源码深度解析 之 submit 方法 (二)

时间:2021-08-12 09:40:08

在上一篇《并发编程(十一)—— Java 线程池 实现原理与源码深度解析(一)》中提到了线程池ThreadPoolExecutor的原理以及它的execute方法。这篇文章是接着上一篇文章写的,如果你没有阅读上一篇文章,建议你去读读。本文解析ThreadPoolExecutor#submit。

  对于一个任务的执行有时我们不需要它返回结果,但是有我们需要它的返回执行结果。对于线程来讲,如果不需要它返回结果则实现Runnable,而如果需要执行结果的话则可以实现Callable。在线程池同样execute提供一个不需要返回结果的任务执行,而对于需要结果返回的则可调用其submit方法。

AbstractExecutorService

我们把上一篇文章的代码贴过来

 public abstract class AbstractExecutorService implements ExecutorService {

     // RunnableFuture 是用于获取执行结果的,我们常用它的子类 FutureTask
// 下面两个 newTaskFor 方法用于将我们的任务包装成 FutureTask 提交到线程池中执行
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
} protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
} // 提交任务
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
// 1. 将任务包装成 FutureTask
RunnableFuture<Void> ftask = newTaskFor(task, null);
// 2. 交给执行器执行,execute 方法由具体的子类来实现
// 前面也说了,FutureTask 间接实现了Runnable 接口。
execute(ftask);
return ftask;
} public <T> Future<T> submit(Runnable task, T result) {
if (task == null) throw new NullPointerException();
// 1. 将任务包装成 FutureTask
RunnableFuture<T> ftask = newTaskFor(task, result);
// 2. 交给执行器执行
execute(ftask);
return ftask;
} public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
// 1. 将任务包装成 FutureTask
RunnableFuture<T> ftask = newTaskFor(task);
// 2. 交给执行器执行
execute(ftask);
return ftask;
}
}

尽管submit方法能提供线程执行的返回值,但只有实现了Callable才会有返回值,而实现Runnable的线程则是没有返回值的,也就是说在上面的3个方法中,submit(Callable<T> task)能获取到它的返回值,submit(Runnable task, T result)能通过传入的载体result间接获得线程的返回值或者准确来说交给线程处理一下,而最后一个方法submit(Runnable task)则是没有返回值的,就算获取它的返回值也是null。

使用示例

submit(Callable<T> task)

 /**
* @author: ChenHao
* @Date: Created in 14:54 2019/1/11
*/
public class Test {
public static void main(String[] args) throws ExecutionException, InterruptedException {
Callable<String> callable = new Callable<String>() {
public String call() throws Exception {
System.out.println("This is ThreadPoolExetor#submit(Callable<T> task) method.");
return "result";
}
}; ExecutorService executor = Executors.newSingleThreadExecutor();
Future<String> future = executor.submit(callable);
executor.shutdown();
System.out.println(future.get());
}
}

运行结果:

并发编程(十二)—— Java 线程池 实现原理与源码深度解析 之 submit 方法 (二)

submit(Runnable task, T result)

 /**
* @author: ChenHao
* @Date: Created in 14:54 2019/1/11
*/
public class Test {
public static void main(String[] args) throws ExecutionException, InterruptedException { ExecutorService executor = Executors.newSingleThreadExecutor();
Data data = new Data();
Future<Data> future = executor.submit(new Task(data), data);
executor.shutdown();
System.out.println(future.get().getName());
}
}
class Data {
String name;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
} class Task implements Runnable {
Data data;
public Task(Data data) {
this.data = data;
}
@Override
public void run() {
System.out.println("This is ThreadPoolExetor#submit(Runnable task, T result) method.");
data.setName("陈浩");
}
}

运行结果:

并发编程(十二)—— Java 线程池 实现原理与源码深度解析 之 submit 方法 (二)

submit(Runnable task)

 /**
* @author: ChenHao
* @Date: Created in 14:54 2019/1/11
*/
public class Test {
public static void main(String[] args) throws ExecutionException, InterruptedException {
Runnable runnable = new Runnable() {
@Override
public void run() {
System.out.println("This is ThreadPoolExetor#submit(Runnable runnable) method.");
}
}; ExecutorService executor = Executors.newSingleThreadExecutor();
Future future = executor.submit(runnable);
executor.shutdown();
System.out.println(future.get());
}
}

运行结果:

并发编程(十二)—— Java 线程池 实现原理与源码深度解析 之 submit 方法 (二)

从上面的源码可以看到,这三者方法几乎是一样的,关键就在于:

 RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);

是如何将一个任务作为参数传递给了newTaskFor,然后调用execute方法,最后进而返回ftask的呢?

 protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
} protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
}

源码分析

这里我建议大家去看看我之前的一篇文章《Java 多线程(五)—— 线程池基础 之 FutureTask源码解析

submit(Callable<T> task)

我们看上面的源码中知道实际上是调用了如下代码

 protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
}

我们看看 FutureTask 的结构

 public class FutureTask<V> implements RunnableFuture<V> {
private volatile int state;
private static final int NEW = 0; //初始状态
private static final int COMPLETING = 1; //结果计算完成或响应中断到赋值给返回值之间的状态。
private static final int NORMAL = 2; //任务正常完成,结果被set
private static final int EXCEPTIONAL = 3; //任务抛出异常
private static final int CANCELLED = 4; //任务已被取消
private static final int INTERRUPTING = 5; //线程中断状态被设置ture,但线程未响应中断
private static final int INTERRUPTED = 6; //线程已被中断 //将要执行的任务
private Callable<V> callable; //用于get()返回的结果,也可能是用于get()方法抛出的异常
private Object outcome; // non-volatile, protected by state reads/writes //执行callable的线程,调用FutureTask.run()方法通过CAS设置
private volatile Thread runner; //栈结构的等待队列,该节点是栈中的最顶层节点。
private volatile WaitNode waiters; public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW; // ensure visibility of callable
}
....
}
 public interface RunnableFuture<V> extends Runnable, Future<V> {
/**
* Sets this Future to the result of its computation
* unless it has been cancelled.
*/
void run();
}

我们知道 FutureTask 继承了 Runnable,这里将 Callable<T> callable 的实例封装成 FutureTask 传给 execute(ftask);我们再来看看上一篇文章中线程运行的代码

 // 此方法由 worker 线程启动后调用,这里用一个 while 循环来不断地从等待队列中获取任务并执行
// 前面说了,worker 在初始化的时候,可以指定 firstTask,那么第一个任务也就可以不需要从队列中获取
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
// 该线程的第一个任务(如果有的话)
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
// 循环调用 getTask 获取任务
while (task != null || (task = getTask()) != null) {
w.lock();
// 如果线程池状态大于等于 STOP,那么意味着该线程也要中断
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
beforeExecute(wt, task);
Throwable thrown = null;
try {
// 到这里终于可以执行任务了,这里是最重要的,task是什么?是Worker 中的firstTask属性 task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
afterExecute(task, thrown);
}
} finally {
// 一个任务执行完了,这个线程还可以复用,接着去队列中拉取任务执行
// 置空 task,准备 getTask 获取下一个任务
task = null;
// 累加完成的任务数
w.completedTasks++;
// 释放掉 worker 的独占锁
w.unlock();
}
}
completedAbruptly = false;
} finally {
// 如果到这里,需要执行线程关闭:
// 说明 getTask 返回 null,也就是超过corePoolSize的线程过了超时时间还没有获取到任务,也就是说,这个 worker 的使命结束了,执行关闭
processWorkerExit(w, completedAbruptly);
}
}

由上面第6行代码 task 就是execute(ftask)传入的任务,第26行 task.run(); 实际上就是 new FutureTask<T>(callable).run(),我们看看FutureTask中的run()方法

 public void run() {
//保证callable任务只被运行一次
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 {
//执行任务,上面的例子我们可以看出,call()里面可能是一个耗时的操作,不过这里是同步的
result = c.call();
//上面的call()是同步的,只有上面的result有了结果才会继续执行
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
setException(ex);
}
if (ran)
//执行完了,设置result
set(result);
}
}
finally {
runner = null;
int s = state;
//判断该任务是否正在响应中断,如果中断没有完成,则等待中断操作完成
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}

在 FutureTask的构造方法中 this.callable = callable; ,因此我们可以知道上面run()方法中第6行 c 就是 代码示例中的 new Callable<String>(),c.call()就是调用 代码示例中new Callable 的call方法,并且这里可以取到返回结果,第22行处设置FutureTask 中 outcome 的值,这样线程就可以取到返回值了。

 protected void set(V v) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}

取值我就不分析了,我之前的文章里面有详细分析。

submit(Runnable task, T result)

 public <T> Future<T> submit(Runnable task, T result) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task, result);
execute(ftask);
return ftask;
} protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
}

我们来看看FutureTask的另外一个构造方法

 public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW; // ensure visibility of callable
}
 public static <T> Callable<T> callable(Runnable task, T result) {
if (task == null)
throw new NullPointerException();
return new RunnableAdapter<T>(task, result);
} static final class RunnableAdapter<T> implements Callable<T> {
final Runnable task;
final T result;
RunnableAdapter(Runnable task, T result) {
this.task = task;
this.result = result;
}
public T call() {
task.run();
return result;
}
}

上面将 runnable, result 封装成了 RunnableAdapter 作为FutureTask的callable属性,这和上面的submit(Callable<T> task) 是不同的,submit(Callable<T> task)是直接将 Callable<T> task作为FutureTask的callable属性。我们看看FutureTask中的run()方法中第6行 c 就是FutureTask 构造方法中的new RunnableAdapter<T>(task, result) ,c.call()就是调用 RunnableAdapter<T>(task, result) 的call方法,call()中的task.run()就是上面代码示例中new Task(data) 中的 run(),run()方法中业务大代码改变了data对象的属性,callable(Runnable task, T result)中也是传的相同的对象data, 所以,result = c.call(); 就是把更改后的data返回,并且将data设置为设置FutureTask 中 outcome 的值,后面的逻辑就是一样的了。

这里可以看成将同一个data传入线程进行处理,同时这个data也传入FutureTask中,并且在RunnableAdapter通过属性进行保存data,等线程将data处理完了,由于是同一个对象,RunnableAdapter中的result也就是data指向的是同一个对象,然后把此result返回到FutureTask保存在属性outcome中,就可以通过FutureTask.get()取到运行结果了。

如果new FutureTask<T>(runnable, null),则result = c.call(); 返回的值也是null,最后从线程池中get的值也是null。