Java并发编程 - Runnbale、Future、Callable 你不知道的那点事(二)

时间:2023-12-17 13:20:44

  Java并发编程 - Runnbale、Future、Callable 你不知道的那点事(一)大致说明了一下 Runnable、Future、Callable 接口之间的关系,也说明了一些内部常用的方法的含义,那具体内部怎么实现的呢?JDK内部底层源码怎么解读?我就带领大家一一探个究竟。

一、ExecutorService 中常用的 submit() 底层源码详解

(1)ExecutorService 中常用的 submit() 方法展示:

<T> Future<T> submit(Callable<T> task);
Future<?> submit(Runnable task);

   暂时只需要知道线程池中的线程是依赖于 ExecutorService 接口进行调用的,实现类是 ThreadPoolExecutor,重写了 Executor 接口的 execute() 方法,然后进行调用;

  后面我会专门讲解一下 Executor、ExecutorService、AbstractExecutorService、ThreadPoolExecutor、Executors 之间的关系,让你详细了解线程池的原理和概念。

(2)ExecutorService 中的 submit(Runnable task) 方法详解

  首先通过 Executors 工具类方法创建线程池(单线程线程池、固定数量线程池、缓存线程池),然后调用 submit() 方法,传入的参数切记是 Runnable 接口的实现类,肯定是重写了 run() 方法;

  明白上面这句话的原理,看一下 AbstractExecutorService 抽象类中的 submit(Runnable task) 方法详解:一步一步调用

public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}

  FutureTask 构造方法,设置参数 callable,state 参数:

public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);    // 调用线程工具类 Executors 中的 callable() 方法
this.state = NEW;
}

  Executors 工具类中的 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;
}
}

  1)先判断 task 不能为空,然后调用 newTaskFor(task, null) 方法,然后一个 RunabbleFuture 接口的实现类实例对象: FuntureTask 实例对象【切记:这个类肯定重写了 run() 方法】;

  2)注意:创建 FutureTask 对象实例时,调用工具类 Executors 创建一个 RunnableAdapter 对象(可以理解为:Runnable 和Callable 之间的适配器),RunnableAdapter 内部类继承了 Callable 接口,重写了 call() 方法,在 call() 方法中调用真正任务要执行的 run() 方法逻辑;

    3)然后调用 execute(ftask) 方法,这个方法是线程池实现顶层接口 Executor,重写了 execute() 方法,去执行任务逻辑方法;

  ExecutorPoolThread 类中的 execute() 方法:

public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false); // 添加到工作线程中
}
else if (!addWorker(command, false))
reject(command);
}

  线程池 ExecutorService 直接调用 execute() 方法执行线程任务,则会创建新的 Worker 对象(Worker 类中有一个 Thread 成员变量),Worker 对象可以理解为一个工作线程,是用HashSet() 集合存储,之后该线程调用 start() 方法线程开始执行任务;

  注意:创建 FutureTask 对象实例时,调用工具类 Executors 创建一个 RunnableAdapter 对象(可以理解为:Runnable 和Callable 之间的适配器),RunnableAdapter 内部类继承了 Callable 接口,重写了 call() 方法,在 call() 方法中调用真正任务要执行的 run() 方法逻辑;

  本质:调用FutureTask类的run( )方法,在其run( )方法中调用了RunnableAdapter类的call( )方法,在call( )方法中调用了Runnable实现类的run( ) 方法!

(3)ExecutorService 中的 submit(Callable task) 方法详解

  实现 Callable 接口的任务,在线程池底层原理调用上是比较简单的,没有 submit(Runnable task) 这么复杂;

  AbstractExecutorService 抽象类中的 submit(Callable task) 方法详解:

public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}

  FutureTask 类中设置 callable、state 参数

protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
} public FutureTask(Callable<V> callable) {
if (callable == null)
throw new NullPointerException();
this.callable = callable;
this.state = NEW; // ensure visibility of callable
}

  线程池 ExecutorService 直接调用 execute() 方法执行线程任务,则会创建新的 Worker 对象(Worker 类中有一个 Thread 成员变量),Worker 对象可以理解为一个工作线程,是用HashSet() 集合存储,之后该线程调用 start() 方法线程开始执行任务;

  本质:在调用 ExecutorPoolThread 中的 run() 方法执行任务时,在里面是调用了真正创建的 FutureTask 实例的 run() 方法,在这里面才真正调用了实现 Callable 接口的 call() 方法,call() 方法中的代码就是真正要执行任务的逻辑代码;

二、实践

  讲了这么多原理实在是太枯燥乏味,一时半会也理解不了,咱就结合实践再加深一下印象!

   1)使用 Callable + Future + ExecutorPoolThread 获取执行结果


public class Test1 {
public static void main(String[] args) {
ExecutorService executor = Executors.newCachedThreadPool();
Task task = new Task();
Future<Integer> result = executor.submit(task);
executor.shutdown();

try {
Thread.sleep(1000);
} catch (InterruptedException e1) {
e1.printStackTrace();
}

System.out.println("主线程执行......");
try {
System.out.println("task运行结果" + result.get());
} catch (Exception e) {
e.printStackTrace();
}
System.out.println("执行完毕");
}
}

class Task implements Callable<Integer> {
@Override
public Integer call() throws Exception {
System.out.println("子线程执行......");
Thread.sleep(3000);
int sum = 0;
for (int i = 0; i < 100; i++)
sum += i;
return sum;
}
}
 

  结果:

子线程执行......
主线程执行......
task运行结果4950
执行完毕

  2)Runnable + ExecutorPoolThread 获取不到结果

public class Test2 {
public static void main(String[] args) {
ExecutorService executor = Executors.newCachedThreadPool();
Task2 task = new Task2();
Future result = executor.submit(task);
executor.shutdown(); try {
Thread.sleep(1000);
} catch (InterruptedException e1) {
e1.printStackTrace();
} System.out.println("主线程执行......");
try {
System.out.println("task运行结果" + result.get());
} catch (Exception e) {
e.printStackTrace();
}
System.out.println("执行完毕");
}
} class Task2 implements Runnable {
@Override
public void run() {
try {
System.out.println("子线程执行......");
Thread.sleep(3000);
Integer sum = 0;
for (int i = 0; i < 100; i++) {
sum += i;
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

  结果:

子线程执行......
主线程执行......
task运行结果null
执行完毕

  task运行结果是 null,由于不是 Callable 接口的实现类,没有通用性,就使用了 RunnableAdapter 适配器来保证通用性,所以在 RunnableAdapter 实现了 Callable 接口中重写了 call() 方法,由于传入的 result 是 null,所以返回的 result 也就是 null 了。

三、总结

  说了这么多Runnable、Future、Callable 底层调用的实现原理,本质都会创建一个 FutureTask 对象,然后调用 run() 方法,最后再调用 call() 方法实现任务逻辑【若是 Runnable 接口的实例对象,会使用 RunnableAdapter 适配器来转接调用一下,然后在调用 call() 方法,在此 call() 方法里面在调用 run() 方法执行;若是 Callable 接口的实例,直接调用 call() 方法执行】

  菜鸟的简单整理,大家仔细阅读,有问题请留言一起讨论讨论!