快捷導(dǎo)航

Java中Future、FutureTask原理以及與線程池的搭配使用

更新時(shí)間：2019年09月24日 10:07:40 作者：wei_lei

這篇文章主要為大家詳細(xì)介紹了Java中Future、FutureTask原理以及與線程池的搭配使用，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下

Java中的Future和Future通常和線程池搭配使用，用來獲取線程池返回執(zhí)行后的返回值。我們假設(shè)通過Executors工廠方法構(gòu)建一個(gè)線程池es ，es要執(zhí)行某個(gè)任務(wù)有兩種方式，一種是執(zhí)行 es.execute(runnable) ,這種情況是沒有返回值的；另外一種情況是執(zhí)行 es.submit(runnale)或者 es.submit(callable) ,這種情況會(huì)返回一個(gè)Future的對(duì)象，然后調(diào)用Future的get()來獲取返回值。

Future

public interface Future<V> {
 boolean cancel(boolean mayInterruptIfRunning);
 
 boolean isCancelled();
 
 boolean isDone();
 
 V get() throws InterruptedException, ExecutionException;
 
 V get(long timeout, TimeUnit unit)
 throws InterruptedException, ExecutionException, TimeoutException;
}

Future是一個(gè)接口，他提供給了我們方法來檢測當(dāng)前的任務(wù)是否已經(jīng)結(jié)束，還可以等待任務(wù)結(jié)束并且拿到一個(gè)結(jié)果，通過調(diào)用Future的get（）方法可以當(dāng)任務(wù)結(jié)束后返回一個(gè)結(jié)果值，如果工作沒有結(jié)束，則會(huì)阻塞當(dāng)前線程，直到任務(wù)執(zhí)行完畢，我們可以通過調(diào)用cancel（）方法來停止一個(gè)任務(wù)，如果任務(wù)已經(jīng)停止，則cancel（）方法會(huì)返回true；如果任務(wù)已經(jīng)完成或者已經(jīng)停止了或者這個(gè)任務(wù)無法停止，則cancel（）會(huì)返回一個(gè)false。當(dāng)一個(gè)任務(wù)被成功停止后，他無法再次執(zhí)行。isDone（）和isCancel（）方法可以判斷當(dāng)前工作是否完成和是否取消。

線程池中有以下幾個(gè)方法：

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(Runnable task, T result) {
 if (task == null) throw new NullPointerException();
 RunnableFuture<T> ftask = newTaskFor(task, result);
 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;
 }

都返回一個(gè)Future對(duì)象，仔細(xì)查看發(fā)現(xiàn)，所有的方法最終都將runnable或者callable轉(zhuǎn)變成一個(gè)RunnableFuture的對(duì)象，這個(gè)RunnableFutre的對(duì)象是一個(gè)同時(shí)繼承了Runnable和Future的接口

public interface RunnableFuture<V> extends Runnable, Future<V> {
 void run();
}

然后調(diào)用executor(runnable)方法，關(guān)于executor(runnable)的具體實(shí)現(xiàn)我們后面再講。最后返回一個(gè)RunnableFuture對(duì)象。RunnableFuture這個(gè)接口直有一個(gè)具體的實(shí)現(xiàn)類，那就時(shí)我們接下來要講的FutureTask。

FutureTask

public class FutureTask<V> implements RunnableFuture<V>

FutureTask實(shí)現(xiàn)了RunnableFuture的接口，既然我們知道最終返回的是一個(gè)FutureTask對(duì)象ftask，而且我們可以通過ftask.get()可以的來得到execute(task)的返回值，這個(gè)過程具體事怎么實(shí)現(xiàn)的了？？這個(gè)也是本篇文章的所要講的。

我們可以先來猜測一下它的實(shí)現(xiàn)過程，首先Runnable的run()是沒有返回值的，所以當(dāng)es.submit()的參數(shù)只有一個(gè)Runnable對(duì)象的時(shí)候，通過ftask.get()得到的也是一個(gè)null值，當(dāng)參數(shù)還有一個(gè)result的時(shí)候，就返回這個(gè)result；如果參數(shù)是一個(gè)Callable的對(duì)象的時(shí)候，Callable的call()是有返回值的，同時(shí)這個(gè)call()方法會(huì)在轉(zhuǎn)換的Runable對(duì)象ftask的run()方法中被調(diào)用，然后將它的返回值賦值給一個(gè)全局變量，最后在ftask的get()方法中得到這個(gè)值。猜想對(duì)不對(duì)了？我們直接看源碼。

將Runnable對(duì)象轉(zhuǎn)為RunnableFuture的方法：

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

public FutureTask(Runnable runnable, V result) {
 this.callable = Executors.callable(runnable, result);
 this.state = NEW; // ensure visibility of 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);
 }

Executors的內(nèi)部類RunnableAdapter

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;
 }
 }

將Callable對(duì)象轉(zhuǎn)為RunnableFuture的方法：

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
 }

接下來我們來看execute(runnbale)的執(zhí)行過程：

execute(runnable)最終的實(shí)現(xiàn)是在ThreadPoolExecutor，基本上所有的線程池都是通過ThreadPoolExecutor的構(gòu)造方法傳入不同的參數(shù)來構(gòu)造的。

ThreadPoolExecutor::executor(runnable) :

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);
 }

如上所示，這個(gè)過程分為三步：

Step 1:

如果當(dāng)前線程池的的線程小于核心線程的數(shù)量的時(shí)候，就會(huì)調(diào)用addWorker檢查運(yùn)行狀態(tài)和正在運(yùn)行的線程數(shù)量，通過返回false來防止錯(cuò)誤地添加線程，然后執(zhí)行當(dāng)前任務(wù)。

Step 2:

否則當(dāng)前線程池的的線程大于核心線程的數(shù)量的時(shí)候，我們?nèi)匀恍枰扰袛嗍欠裥枰砑右粋€(gè)新的線程來執(zhí)行這個(gè)任務(wù)，因?yàn)榭赡芤呀?jīng)存在的線程此刻任務(wù)執(zhí)行完畢處于空閑狀態(tài)，這個(gè)時(shí)候可以直接復(fù)用。否則創(chuàng)建一個(gè)新的線程來執(zhí)行此任務(wù)。

Step 3:

如果不能再添加新的任務(wù)，就拒絕。

執(zhí)行execute(runnable)最終會(huì)回調(diào)runnable的run()方法，也就是FutureTask的對(duì)象ftask的run()方法，源碼如下：

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)
   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);
 }
 }

通過執(zhí)行result = c.call()拿到返回值，然后set(result) ，因此get()方法獲得的值正是這個(gè)result。

以上就是本文的全部內(nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: