[Java] Java8 自定义CompletableFuture的原理解析

• Java8 自定义CompletableFuture原理
  
• CompleteFuture简单使用
  
  
  
  • 下面简单介绍用法
    
  
  

public class FutureInAction3 {
  public static void main(String[] args) {
    Future<String> future = invoke(() -> {
      try {
        Thread.sleep(10000L);
        return "I am Finished.";
      } catch (InterruptedException e) {
        return "I am Error";
      }
    });
    future.setCompletable(new Completable<String>() {
      @Override
      public void complete(String s) {
        System.out.println("complete called ---- " + s);
      }
      @Override
      public void exception(Throwable cause) {
        System.out.println("error");
        cause.printStackTrace();
      }
    });
    System.out.println("....do something else .....");
    System.out.println("try to get result ->" + future.get());
  }
  private static <T> Future<T> invoke(Callable<T> callable) {
    AtomicReference<T> result = new AtomicReference<>();
    AtomicBoolean finished = new AtomicBoolean(false);
    Future<T> future = new Future<T>() {
      private Completable<T> completable;
      @Override
      public T get() {
        return result.get();
      }
      @Override
      public boolean isDone() {
        return finished.get();
      }
      // 设置完成
      @Override
      public void setCompletable(Completable<T> completable) {
        this.completable = completable;
      }
      // 获取
      @Override
      public Completable<T> getCompletable() {
        return completable;
      }
    };
    Thread t = new Thread(() -> {
      try {
        T value = callable.action();
        result.set(value);
        finished.set(true);
        if (future.getCompletable() != null)
          future.getCompletable().complete(value);
      } catch (Throwable cause) {
        if (future.getCompletable() != null)
          future.getCompletable().exception(cause);
      }
    });
    t.start();
    return future;
  }
  private interface Future<T> {
    T get();
    boolean isDone();
    //  1
    void setCompletable(Completable<T> completable);
    //  2
    Completable<T> getCompletable();
  }
  private interface Callable<T> {
    T action();
  }
  // 回调接口
  private interface Completable<T> {
    void complete(T t);
    void exception(Throwable cause);
  }
}

package demos;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
/**
 * @author djh on  2019/4/22 10:23
 * @E-Mail 1544579459@qq.com
 */
public class Demo {
  public static void main(String[] args) throws ExecutionException, InterruptedException {
    ExecutorService cachePool = Executors.newCachedThreadPool();
    Future<String> future = cachePool.submit(() -> {
      Thread.sleep(3000);
      return "异步任务计算结果!";
    });
    // 提交完异步任务后, 主线程可以继续干一些其他的事情.
    doSomeThingElse();
    // 为了获取异步计算结果, 我们可以通过 future.get 和 轮询机制来获取.
    String result;
    // Get 方式会导致当前线程阻塞, 这显然违背了异步计算的初衷.
    // result = future.get();
    // 轮询方式虽然不会导致当前线程阻塞, 但是会导致高额的 CPU 负载.
    long start = System.currentTimeMillis();
    while (true) {
      if (future.isDone()) {
        break;
      }
    }
    System.out.println("轮询耗时:" + (System.currentTimeMillis() - start));    
    result = future.get();
    System.out.println("获取到异步计算结果啦: " + result);
    cachePool.shutdown();
  }
  private static void doSomeThingElse() {
    try {
      Thread.sleep(1000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    System.out.println("我的最重要的事情干完了, 我要获取异步计算结果来执行剩下的事情.");
  }
}

我的最重要的事情干完了, 我要获取异步计算结果来执行剩下的事情.
轮询耗时:2000
获取到异步计算结果啦: 异步任务计算结果!
Process finished with exit code 0

package demos;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
 * @author djh on  2019/5/1 20:26
 * @E-Mail 1544579459@qq.com
 */
public class CompleteFutureDemo {
  public static void main(String[] args) throws ExecutionException, InterruptedException {
    CompletableFuture<String> completableFutureOne = new CompletableFuture<>();
    ExecutorService cachePool = Executors.newCachedThreadPool();
    cachePool.execute(() -> {
      try {
        Thread.sleep(3000);
        completableFutureOne.complete("异步任务执行结果");
        System.out.println(Thread.currentThread().getName());
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    });
    // WhenComplete 方法返回的 CompletableFuture 仍然是原来的 CompletableFuture 计算结果.
    CompletableFuture<String> completableFutureTwo = completableFutureOne.whenComplete((s, throwable) -> {
      System.out.println("当异步任务执行完毕时打印异步任务的执行结果: " + s);
    });
    // ThenApply 方法返回的是一个新的 completeFuture.
    CompletableFuture<Integer> completableFutureThree = completableFutureTwo.thenApply(s -> {
      System.out.println("当异步任务执行结束时, 根据上一次的异步任务结果, 继续开始一个新的异步任务!");
      try {
        Thread.sleep(1000);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
      return s.length();
    });
    System.out.println("阻塞方式获取执行结果:" + completableFutureThree.get());
    cachePool.shutdown();
  }
}