快捷導(dǎo)航

Java異步編程Future應(yīng)用方式

更新時(shí)間：2025年02月17日 08:39:40 作者：趙廣陸

Java中的Future接口用于構(gòu)建復(fù)雜并行操作,它允許異步執(zhí)行任務(wù),并在需要時(shí)獲取結(jié)果,通過Future接口,可以避免多線程編程中的一些常見問題,如線程執(zhí)行順序和結(jié)果獲取的復(fù)雜性,然而,在使用Future時(shí)需要注意,并行執(zhí)行可能會變?yōu)榇袌?zhí)行,特別是在使用get()方法時(shí)

1 Future接口介紹

此時(shí)有的人會說，對于任務(wù)并行需求，直接通過多線程實(shí)現(xiàn)不就可以了，要注意，對于多線程的實(shí)現(xiàn)，java提供了三種方式：繼承Thread類、實(shí)現(xiàn)Runnable接口和實(shí)現(xiàn)Callable接口。

但是業(yè)務(wù)代碼在執(zhí)行時(shí)會考慮執(zhí)行順序的問題，直接基于這些方式實(shí)現(xiàn)多線程會出現(xiàn)兩個(gè)問題：

1）要想控制線程執(zhí)行順序，會通過join()等待線程結(jié)束，那這樣的話又回歸到了阻塞式調(diào)用的思路上，違背了并行的需求。另外還可以通過wait()、notify()、notifyAll()結(jié)合狀態(tài)變量實(shí)現(xiàn)，但實(shí)現(xiàn)起來過于復(fù)雜。
2）線程執(zhí)行完之后，要想獲取線程執(zhí)行結(jié)果，還要用過共享變量或線程間通信等方式來獲取，同樣過于復(fù)雜。為了解決上述問題，Java5中推出了Future，其初衷就是用于構(gòu)建復(fù)雜并行操作。內(nèi)部方法在返回時(shí)，不是返回一個(gè)值，而是返回Future對象。其本質(zhì)是在執(zhí)行主業(yè)務(wù)的同時(shí)，異步的執(zhí)行其他分業(yè)務(wù)，從而利用原本需要同步執(zhí)行時(shí)的等待時(shí)間去執(zhí)行其他的業(yè)務(wù)，當(dāng)需要獲取其結(jié)果時(shí)，再進(jìn)行獲取。

Java官網(wǎng)對于Future的描述：

Future表示異步計(jì)算的結(jié)果。提供了一些方法來檢查計(jì)算是否完成，等待其完成以及檢索計(jì)算結(jié)果。只有在計(jì)算完成后才可以使用get方法檢索結(jié)果，必要時(shí)將其阻塞，直到準(zhǔn)備就緒為止。取消通過cancel方法執(zhí)行。提供了其他方法來確定任務(wù)是正常完成還是被取消。一旦計(jì)算完成，就不能取消計(jì)算。

在Future接口中有五個(gè)抽象方法：

cancel()：取消任務(wù), 取消成功返回true；入?yún)ayInterruptIfRunning表示是否允許取消正在執(zhí)行中的任務(wù)。

isCancelled()：返回布爾值，代表是否取消成功。

isDone()：返回布爾值，代表是否執(zhí)行完畢。

get()：返回Future對象，獲取執(zhí)行結(jié)果，如果任務(wù)沒有完成會阻塞到任務(wù)完成再返回。

2 Future應(yīng)用

Future的使用通常需要配合ExecutorService和Callable一起

使用，使用示例如下：

public class FutureAsyncDemo {
  static Random random = new Random();
  static ExecutorService executor =
Executors.newCachedThreadPool();
  //接收文章名稱，獲取并計(jì)算文章分?jǐn)?shù)
  public static int getArticleScore(String
aname){
    Future<Integer> futureA =
executor.submit(new
CalculateArticleScoreA());
    Future<Integer> futureB =
executor.submit(new
CalculateArticleScoreA());
    Future<Integer> futureC =
executor.submit(new
CalculateArticleScoreA());
    doSomeThingElse();
    Integer a = null;
    try {
      a = futureA.get();
   } catch (InterruptedException e) {
      futureA.cancel(true);
      e.printStackTrace();
   } catch (ExecutionException e) {
      futureA.cancel(true);
       e.printStackTrace();
   }
    Integer b = null;
    try {
      b = futureB.get();
   } catch (InterruptedException e) {
      futureB.cancel(true);
      e.printStackTrace();
   } catch (ExecutionException e) {
      futureB.cancel(true);
      e.printStackTrace();
   }
    Integer c = null;
    try {
      c = futureC.get();
   } catch (InterruptedException e) {
      futureC.cancel(true);
      e.printStackTrace();
   } catch (ExecutionException e) {
      futureC.cancel(true);
      e.printStackTrace();
   }
    executor.shutdown();
    return a+b+c;
 }
  private static void doSomeThingElse() {
    System.out.println("exec other
things");
 }
  public static void main(String[] args) {
  
 System.out.println(getArticleScore("demo"))
;
 }
}
class CalculateArticleScoreA implements
Callable<Integer>{
  @Override
  public Integer call() throws Exception {
    //業(yè)務(wù)代碼
    Random random = new Random();
    TimeUnit.SECONDS.sleep(3);
  
 System.out.println(Thread.currentThread().g
etName());
    return random.nextInt(100);
 }
}

執(zhí)行結(jié)果

exec other things
pool-1-thread-1
pool-1-thread-3
pool-1-thread-2
159

上述方法改造了calculateArticleScore()，在其內(nèi)部基于線程池調(diào)用重寫了Callable接口中的call()，并在call()中對具體業(yè)務(wù)完成編碼，并且讓其在執(zhí)行時(shí)睡三秒鐘。根據(jù)結(jié)果可以看到，先調(diào)用了計(jì)算文章分?jǐn)?shù)方法，其內(nèi)部開啟了子線程去執(zhí)行任務(wù)，并且子線程在執(zhí)行時(shí)，并沒有阻塞主線程的執(zhí)行。

當(dāng)主線程需要結(jié)果時(shí)，在通過返回的Future來獲取子任務(wù)中的返回值。

3 Future并行變串行問題解析

剛才已經(jīng)基于Future演示了并行執(zhí)行的效果，已經(jīng)達(dá)到了期望，但是在使用的過程中，其實(shí)還有個(gè)坑需要說明。

對于Future的使用，如稍加不注意，就會讓并行變?yōu)榇小?/p>

示例代碼如下：

public class FutureAsyncDemo {
  static ExecutorService executor =
Executors.newCachedThreadPool();
  //接收文章名稱，獲取并計(jì)算文章分?jǐn)?shù)
  public static int getArticleScore(String
aname){
    Future<Integer> futureA =
executor.submit(new
CalculateArticleScoreA());
    Future<Integer> futureB =
executor.submit(new
CalculateArticleScoreB());
     Future<Integer> futureC =
executor.submit(new
CalculateArticleScoreC());
    doSomeThingElse();
    Integer a = 0;
    try {
      a = futureA.get();
   } catch (InterruptedException e) {
      futureA.cancel(true);
      e.printStackTrace();
   } catch (ExecutionException e) {
      futureA.cancel(true);
      e.printStackTrace();
   }
    Integer b = 0;
    try {
      b = futureB.get();
   } catch (InterruptedException e) {
      futureB.cancel(true);
      e.printStackTrace();
   } catch (ExecutionException e) {
      futureB.cancel(true);
      e.printStackTrace();
   }
    Integer c = 0;
    try {
      c = futureC.get();
   } catch (InterruptedException e) {
       futureC.cancel(true);
      e.printStackTrace();
   } catch (ExecutionException e) {
      futureC.cancel(true);
      e.printStackTrace();
   }
    executor.shutdown();
    return a+b+c;
 }
  private static void doSomeThingElse() {
    System.out.println("exec other
things");
 }
  public static void main(String[] args) {
  
 System.out.println(getArticleScore("demo"))
;
 }
}
class CalculateArticleScoreA implements
Callable<Integer>{
  @Override
  public Integer call() throws Exception {
    Random random = new Random();
    TimeUnit.SECONDS.sleep(10);
   
 System.out.println(Thread.currentThread().g
etName());
    return random.nextInt(100);
 }
}
class CalculateArticleScoreB implements
Callable<Integer>{
  @Override
  public Integer call() throws Exception {
    Random random = new Random();
    TimeUnit.SECONDS.sleep(20);
  
 System.out.println(Thread.currentThread().g
etName());
    return random.nextInt(100);
 }
}
class CalculateArticleScoreC implements
Callable<Integer>{
  @Override
  public Integer call() throws Exception {
    Random random = new Random();
    TimeUnit.SECONDS.sleep(30);
  
 System.out.println(Thread.currentThread().g
etName());
    return random.nextInt(100);
 }
 }

上述代碼加計(jì)算得分方法復(fù)制出來兩份，各自休眠10秒、20秒、30秒。當(dāng)方法返回Future之后，調(diào)用get()進(jìn)行值獲取時(shí)，發(fā)現(xiàn)每次調(diào)用時(shí)都需要進(jìn)行等待。

這樣可以發(fā)現(xiàn)，之前的并行現(xiàn)在變成了串行了！?。。?這個(gè)問題為什么會產(chǎn)生呢？需要看一下Future中對于get()的介紹

根據(jù)源碼可知，當(dāng)調(diào)用get()時(shí)，其會等待對應(yīng)方法執(zhí)行完畢后，才會返回結(jié)果，否則會一直等待。因?yàn)檫@個(gè)設(shè)定，所以上述代碼則出現(xiàn)并行變串行的效果。

對于這個(gè)問題的解決，可以調(diào)用get()的重載，get(longtimeout, TimeUnit unit)。設(shè)置等待的時(shí)長，如果超時(shí)則拋出TimeoutException。

使用示例如下：

public class FutureAsyncDemo {
  static Random random = new Random();
  static ExecutorService executor =
Executors.newCachedThreadPool();
  //接收文章名稱，獲取并計(jì)算文章分?jǐn)?shù)
   public static int
getArticleScore(String aname){
    Future<Integer> futureA =
executor.submit(new
CalculateArticleScoreA());
    Future<Integer> futureB =
executor.submit(new
CalculateArticleScoreB());
    Future<Integer> futureC =
executor.submit(new
CalculateArticleScoreC());
    doSomeThingElse();
    Integer a = 0;
    try {
      a = futureA.get();
   } catch (InterruptedException e) {
      futureA.cancel(true);
      e.printStackTrace();
   } catch (ExecutionException e) {
      futureA.cancel(true);
      e.printStackTrace();
   }
    Integer b = 0;
    try {
       b = futureB.get(3L,
TimeUnit.SECONDS);
   } catch (TimeoutException e) {
      e.printStackTrace();
   }
    catch (InterruptedException e) {
      futureB.cancel(true);
      e.printStackTrace();
   } catch (ExecutionException e) {
      futureB.cancel(true);
      e.printStackTrace();
   }
    Integer c = 0;
    try {
      c = futureC.get();
   } catch (InterruptedException e) {
      futureC.cancel(true);
      e.printStackTrace();
   } catch (ExecutionException e) {
      futureC.cancel(true);
      e.printStackTrace();
   }
    executor.shutdown();
    return a+b+c;
 }
  private static void doSomeThingElse() {
     System.out.println("exec other
things");
 }
  public static void main(String[] args)
{
  
 System.out.println(getArticleScore("demo")
);
 }
}
class CalculateArticleScoreA implements
Callable<Integer>{
  @Override
  public Integer call() throws Exception
{
    Random random = new Random();
    TimeUnit.SECONDS.sleep(10);
  
 System.out.println(Thread.currentThread().
getName());
    return random.nextInt(100);
 }
}
class CalculateArticleScoreB implements
Callable<Integer>{
  @Override
  public Integer call() throws Exception
{
    Random random = new Random();
    TimeUnit.SECONDS.sleep(20);
  
 System.out.println(Thread.currentThread().
getName());
    return random.nextInt(100);
 }
}
class CalculateArticleScoreC implements
Callable<Integer>{
  @Override
  public Integer call() throws Exception
{
    Random random = new Random();
    TimeUnit.SECONDS.sleep(30);
  
 System.out.println(Thread.currentThread().
getName());
    return random.nextInt(100);
 }
}

在上述方法中，對于B的get()設(shè)置了超時(shí)時(shí)間三秒鐘，如果當(dāng)調(diào)用其獲取返回值時(shí)，如果超過三秒仍然沒有返回結(jié)果，則拋出超時(shí)異常，接著方法會再次向下運(yùn)行。

對于Future來說，它能夠支持任務(wù)并發(fā)執(zhí)行，對于任務(wù)結(jié)果的獲取順序是按照提交的順序獲取，在使用的過程中建議通過CPU高速輪詢的方式獲取任務(wù)結(jié)果，但這種方式比較耗費(fèi)資源。不建議使用