ThreadPoolExecutor中的submit()方法详细讲解
作者:张行之 时间:2022-02-18 03:02:39
在使用线程池的时候,发现除了execute()方法可以执行任务外,还发现有一个方法submit()可以执行任务。
submit()有3个参数不一的方法,这些方法都是在ExecutorService接口中声明的,在AbstractExecutorService中实现,而ThreadPoolExecutor继承AbstractExecutorService。
<T> Future<T> submit(Callable<T> callable);
<T> Future<T> submit(Runnable var1, T result);
Future<?> submit(Runnable runnable);
我们可以看到submit()的参数既可以是Runnable,又可以是Callable。对于Runnable我们是比较熟的,它是线程Thread所执行的任务,里面有一个run()方法,是任务的具体执行操作。那么Callable呢?我们一起看下他们的代码吧。
public interface Runnable {
? ? void run();
}
public interface Callable<V> {
? ? V call() throws Exception;
}
Runnable这里就不介绍了,Callable接口定义了一个call()方法,返回一个Callable指定的泛型类,并且call()调用的时候会抛出异常。通过比较Runnable和Callable还看不什么端倪,那么我们就看看内部实现吧。
submmit()参数解析
这里重点分析submit()带参数Runnable和Callable的方法
public Future<?> submit(Runnable task) {
? ? if (task == null) throw new NullPointerException();
? ? RunnableFuture<Void> ftask = newTaskFor(task, null);
? ? execute(ftask);
? ? return ftask;
}
public <T> Future<T> submit(Callable<T> task) {
? ? if (task == null) throw new NullPointerException();
? ? RunnableFuture<T> ftask = newTaskFor(task);
? ? execute(ftask);
? ? return ftask;
}
我们发现2者的实现没有任何的差异,唯一就是submit()参数不同。
参数传入newTaskFor()方法,那么可以肯定就是在这个方法里做了什么操作。
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);
}
newTaskFor()的目的就是创建一个FutureTask对象,那我们追踪到FutureTask的构造方法(FutureTask非常关键,后面会分析)。
public FutureTask(Runnable runnable, V result) {
? ? this.callable = Executors.callable(runnable, result);
? ? this.state = NEW; ? ? ??
}
public FutureTask(Callable<V> callable) {
? ? if (callable == null)throw new NullPointerException();
? ? this.callable = callable;
? ? this.state = NEW; ? ? ??
}
到了这里我们知道,其实Runnable会在这里转化成Callable。我们来看下Executors.callable()具体实现。
public static <T> Callable<T> callable(Runnable task, T result) {
? ? if (task == null)
? ? ? ? throw new NullPointerException();
? ? return new RunnableAdapter<T>(task, result);
}
private static final class RunnableAdapter<T> implements Callable<T> {
? ? private final Runnable task;
? ? private final T result;
? ? RunnableAdapter(Runnable task, T result) {
? ? ? ? this.task = task;
? ? ? ? this.result = result;
? ? }
? ? public T call() {
? ? ? ? task.run();
? ? ? ? return result;
? ? }
}
Executors.callable()创建了一个RunnableAdapter对象,RunnableAdapter实现了Callable接口,在call()方法中调用了传入的Runnable的run(),并且将传入的result参数返回。
也就是说我们调用submit()传入的Runnbale最终会转化成Callable,并且返回一个result值(如果我们传入这个参数则返回这个参数,不传入则返回null)。
到这里我们讲清楚了submit()的参数的区别和内部实现,submit()方法有一个返回值Future,下面我们来分析一下返回值Future。
submit()的返回值Future
上面分析submit()源码可知,submit()返回的是一个RunnableFuture类对象,真正是通过newTaskFor()方法返回一个new FutureTask()对象。所以submit()返回的真正的对象是FutureTask对象。
那么FutureTask是什么,我们来看下它的类继承关系。
public class FutureTask<V> implements RunnableFuture<V> {
? ? ...
}
public interface RunnableFuture<V> extends Runnable, Future<V> {
? ? void run();
}
通过继承关系我们可以明确的知道其实FutureTask就是一个Runnable。并且有自己run()实现。我们来看下FutureTask的run()是如何实现的。
public void run() {
if (state != NEW ||
!U.compareAndSwapObject(this, RUNNER, 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)
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);
}
}
我们在new FutureTask()对象的时候,在FutureTask构造方法中会对state状态赋值为NEW,并且传入一个callable对象。通过FutureTask的run()我们可以知道,其实就通过state状态判断,调用callable的call()。(如果传入的参数是Runnable,Runnable在RunnableAdapter类中转化时,在call()中,其实调用的就是Runnable的run()方法)。
所以在submit()方法中,调用了一个execute(task)的方法,实际执行的是FutureTask的run(),而FutureTask的run()调用的是Callable的call()方法。
说了这么多,submit()最后执行的还是传入的Runnable的run()或Callable的call()方法。好像没有FutureTask什么事啊。
其实不是,submit()返回FutureTask对象,通过这个FutureTask对象调用get()可以返回submit()方法传入的一个泛型类参数result对象,如果是Callable直接通过call()返回。这个返回值的可以用来校验任务执行是否成功。
FutureTask的get()的实现
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;
? ? if (s >= CANCELLED)
? ? ? ? throw new CancellationException();
? ? throw new ExecutionException((Throwable)x);
}
最后是通过outcome参数将根据任务的状态将结果返回。那么outcome参数在哪里赋值了?outcome参数赋值的地方有好2处,一是FutureTask的set(),二是FutureTask的setException()。
set()是在FutureTask的run()执行完成后,将传入的result参数赋值给传入给set(),赋值给outcome参数。如果run()报异常了会将Throwable对象通过setException()方法传入,赋值给outcome变量
大家可以返回上面的run()查看下。
protected void set(V v) {
? ? if (U.compareAndSwapInt(this, STATE, NEW, COMPLETING)) {
? ? ? ? outcome = v;
? ? ? ? U.putOrderedInt(this, STATE, NORMAL); // final state
? ? ? ? finishCompletion();
? ? }
}
protected void setException(Throwable t) {
? ? if (U.compareAndSwapInt(this, STATE, NEW, COMPLETING)) {
? ? ? ? outcome = t;
? ? ? ? U.putOrderedInt(this, STATE, EXCEPTIONAL); // final state
? ? ? ? finishCompletion();
? ? }
}
submit()使用案例
public class Test {
? ? private static final String SUCCESS = "success";
? ? public static void main(String[] args) {
? ? ? ? ExecutorService executorService = Executors.newFixedThreadPool(3);
? ? ? ? System.out.println("------------------任务开始执行---------------------");
? ? ? ? Future<String> future = executorService.submit(new Callable<String>() {
? ? ? ? ? ? @Override
? ? ? ? ? ? public String call() throws Exception {
? ? ? ? ? ? ? ? Thread.sleep(5000);
? ? ? ? ? ? ? ? System.out.println("submit方法执行任务完成" + " ? thread name: " + Thread.currentThread().getName());
? ? ? ? ? ? ? ? return SUCCESS;
? ? ? ? ? ? }
? ? ? ? });
? ? ? ? try {
? ? ? ? ? ? String s = future.get();
? ? ? ? ? ? if (SUCCESS.equals(s)) {
? ? ? ? ? ? ? ? String name = Thread.currentThread().getName();
? ? ? ? ? ? ? ? System.out.println("经过返回值比较,submit方法执行任务成功 ? ?thread name: " + name);
? ? ? ? ? ? }
? ? ? ? } catch (InterruptedException e) {
? ? ? ? ? ? e.printStackTrace();
? ? ? ? } catch (ExecutionException e) {
? ? ? ? ? ? e.printStackTrace();
? ? ? ? }
? ? ? ? System.out.println("-------------------main thread end---------------------");
? ? }
}
打印结果:
------------------任务开始执行---------------------
call()调用开始: 1496899867882
submit方法执行任务完成: 1496899872897 thread name: pool-1-thread-1
经过返回值比较,submit方法执行任务成功 thread name: main
-------------------main thread end---------------------
主线程会一直阻塞,等待线程池中的任务执行完后,在执行后面的语句。
来源:https://blog.csdn.net/qq_33689414/article/details/72955253