1 简介

　　Futrue可以监视目标线程调用call的情况，当你调用Future的get()方法以获得结果时，调用方的线程就被阻塞，直到目标线程的call方法结束并返回结果。

　　线程的实现方式有几种方式，继承Thread类，实现Runnable接口，线程池，callable这种方式。

　　callable和Runnable的区别是callable可以有返回值，也可以抛出异常的特性，而Runnable没有。

　　注意callable可以有返回值，也可以抛出异常这点很关键。
　　很多时候我们让多线程去帮我们处理事情，是需要拿到返回值的，有了异常也可以处理，比如某宝的APP页面，一个页面展示3个块，而每个块展示的信息从后端获取的接口都不一样，那么是让前端调后端3次接口吗？
　　后端可以把3个块的信息，包装成一个接口，全部返回，那么问题来了，后端调用3个接口，比如第一个接口需要1秒，第二个需要2秒，第三个需要3秒，那么包装的这个接口响应时间最少6秒，怎么解决这个问题呢，可以用多线程来帮我们解决。
　　启动3个线程，每个线程去调用一个接口，那么3个线程一共执行完的时间就是最慢的那个线程的执行时间，这样接口的响应时间就变成了3秒，一下节省了一半的时间。
　　那么问题来了，线程如何把执行的业务代码的结果返回来呢？这时候就用到callable了

2 Future

2.1 源码

//它定义了对线程Callable执行的管理，包括执行过程中取消，判断是否取消了，是否执行完成，获取放回结果

public interface Future<V> {

　　//取消Callable的执行，当Callable还没有完成时

    boolean cancel(boolean mayInterruptIfRunning);

　　//是否取消了

    boolean isCancelled();

　　//是否执行完成了

    boolean isDone();

　　//获取返回结果

    V get() throws InterruptedException, ExecutionException;

　　//获取返回结果，若超过指定时间还没有获取到(还在执行中)，直接不获取了

    V get(long timeout, TimeUnit unit)

        throws InterruptedException, ExecutionException, TimeoutException;

}

FutureTask 实现了Future

//FutureTask 的构造函数，传入一个Callable

public FutureTask(Callable<V> callable) {

        if (callable == null)

            throw new NullPointerException();

        this.callable = callable;

        this.state = NEW;       // ensure visibility of callable

    }

2.2 示例1

使用FutureTask 来执行一个线程任务

    public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {

        System.out.println("我在写作业");

        FutureTask f = new FutureTask(()->{

            System.out.println( "发现笔快没油了，叫" + Thread.currentThread().getName() + "帮我去买笔【是否是守护线程" + Thread.currentThread().isDaemon() + "】");

            Thread.sleep(2000);

            System.out.println(Thread.currentThread().getName() + "买回来了");

            return "晨光牌中性笔";

        });

        Thread t = new Thread(f,"弟弟");

        t.start();

        System.out.println("奋笔疾书中aaa");

    }

}

执行结果，看到是主线程执行完成了，而FutureTask异步线程还在继续执行。异步线程是用户线程

我在写作业

奋笔疾书中aaa

发现笔快没油了，叫弟弟帮我去买笔【是否是守护线程false】

弟弟买回来了

2.2 示例2

　　使用FutureTask 来执行一个线程任务，并且获得执行结果

　　我写作业也，发现快没有了，叫弟弟去给我买笔，我继续写作业，等弟弟买回来了，我换上新买的笔继续写作业。

　　达到在写作业的同时去买笔，还可以拿到新买的笔的效果

public class FutureTest {

    public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {

        System.out.println("我在写作业");

        FutureTask f = new FutureTask(()->{

            System.out.println( "发现笔快没油了，叫" + Thread.currentThread().getName() + "帮我去买笔");

            Thread.sleep(2000);

            System.out.println(Thread.currentThread().getName() + "买回来了");

            return "晨光牌中性笔";

        });

        Thread t = new Thread(f,"弟弟");

        t.start();

        System.out.println("奋笔疾书中aaa");

        Object o = f.get();

        System.out.println("奋笔疾书中bbb");

        System.out.println("换弟弟给我新买的：" + o + "写作业");

    }

}

执行结果，发现在Object o = f.get();出发生阻塞了，get()后面的代码，主线程是等待FutureTask异步线程执行完成后继续执行的，所以说get方法会造成阻塞

我在写作业

奋笔疾书中aaa

发现笔快没油了，叫弟弟帮我去买笔

弟弟买回来了

奋笔疾书中bbb

换弟弟给我新买的：晨光牌中性笔写作业

2.3 示例2(不见不散)

　　get方法会找出阻塞

　　把Thread.sleep(500);时间变为5000，再次执行

　　清楚的发现System.out.println("奋笔疾书中bbb");这条语句被阻塞了，它是等笔买回来了，才执行的。也就是f.get()这个方法会造成阻塞

2.4 示例3(过时不候)

　　get方法传入等待时间

public class FutureTest {

    public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {

        System.out.println("我在写作业");

        FutureTask f = new FutureTask(()->{

            System.out.println( "发现笔快没油了，叫" + Thread.currentThread().getName() + "帮我去买笔");

            Thread.sleep(5000);

            System.out.println(Thread.currentThread().getName() + "买回来了");

            return "晨光牌中性笔";

        });

        Thread t = new Thread(f,"弟弟");

        t.start();

        System.out.println("奋笔疾书中aaa");

        Object o = f.get(2, TimeUnit.SECONDS);

        System.out.println("奋笔疾书中bbb");

        System.out.println("换弟弟给我新买的：" + o + "写作业");

    }

}

执行结果，超过2秒，没有获取到结果，主线程直接报错

我在写作业

奋笔疾书中aaa

发现笔快没油了，叫弟弟帮我去买笔

Exception in thread "main" java.util.concurrent.TimeoutException

    at java.util.concurrent.FutureTask.get(FutureTask.java:205)

    at com.ruoyi.weixin.user.SuoTest.FutureTest.main(FutureTest.java:28)

弟弟买回来了

2.5 示例4

　　使用isDone判断是否完成，这样子线程不会阻塞(可以先做点别的)，虽然也是需要等待执行完成才能向下继续执行

public class FutureTest {

    public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {

        System.out.println("我在写作业");

        FutureTask f = new FutureTask(()->{

            System.out.println( "发现笔快没油了，叫" + Thread.currentThread().getName() + "帮我去买笔");

            Thread.sleep(5000);

            System.out.println(Thread.currentThread().getName() + "买回来了");

            return "晨光牌中性笔";

        });

        Thread t = new Thread(f,"弟弟");

        t.start();

        System.out.println("奋笔疾书中aaa");

        while (true){

            if(f.isDone()){

                Object o = f.get();

                System.out.println("换弟弟给我新买的：" + o + "写作业");

                break;

            }else{

                System.out.println("看了眼门口，弟弟还没回来，继续写作业");

                Thread.sleep(500);

            }

        }

    }

}

执行结果

我在写作业

奋笔疾书中aaa

看了眼门口，弟弟还没回来，继续写作业

发现笔快没油了，叫弟弟帮我去买笔

看了眼门口，弟弟还没回来，继续写作业

看了眼门口，弟弟还没回来，继续写作业

看了眼门口，弟弟还没回来，继续写作业

看了眼门口，弟弟还没回来，继续写作业

看了眼门口，弟弟还没回来，继续写作业

看了眼门口，弟弟还没回来，继续写作业

看了眼门口，弟弟还没回来，继续写作业

看了眼门口，弟弟还没回来，继续写作业

看了眼门口，弟弟还没回来，继续写作业

弟弟买回来了

换弟弟给我新买的：晨光牌中性笔写作业

Process finished with exit code 0

2.6 小结

　　1）通过FutureTask 可以创建异步任务，并且可以获得执行结果，而且可以进行异常处理

　　2）FutureTask异步线程是用户线程，不会随着主线程的结束而结束

　　3）调用get方法会造成阻塞(是一个比较大的缺点)

　　4）get(等待时间)，超时了，主线程直接报错

　　5）可以通过isDone方法判断异步任务是否完成

3 CompletableFuture

3.1 简介

　　 CompletableFuture是FutureTas的升级版，FutureTask能做的它能做，FutureTask不能做的它也能做。

　　 CompletableFuture实现了CompletionStage接口和Future接口，增加了异步回调、流式处理、多个Future组合处理的能力，使Java在处理多任务的协同工作时更加顺畅便利

3.2 继承关系图

3.3 异步任务方法supplyAsync / runAsync

　　supplyAsync表示创建带返回值的异步任务的，runAsync表示创建无返回值的异步任务

3.3.1 示例supplyAsync

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {

        System.out.println(Thread.currentThread().getName() + "执行开始");

        CompletableFuture.supplyAsync(()->{

            System.out.println(Thread.currentThread().getName() + "执行1");

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

            try {

                Thread.sleep(2000);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            System.out.println(Thread.currentThread().getName() + "任务执行结束");

            return 1;

        });

        System.out.println(Thread.currentThread().getName() + "执行结束");

    }

执行结果，System.out.println(Thread.currentThread().getName() + "任务执行结束");这一句话没有打印，因为CompletableFuture.supplyAsync异步线程是守护线程，主线程执行结束后，守护线程会立即结束。

main执行开始

main执行结束

ForkJoinPool.commonPool-worker-1执行1

ForkJoinPool.commonPool-worker-1是否是守护线程-true

3.3.2 runAsync方法

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {

        System.out.println(Thread.currentThread().getName() + "执行开始");

        CompletableFuture.runAsync(()->{

            System.out.println(Thread.currentThread().getName() + "执行1");

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

            try {

                Thread.sleep(2000);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            System.out.println(Thread.currentThread().getName() + "任务执行结束");

        });

        System.out.println(Thread.currentThread().getName() + "执行结束");

    }

执行结果，System.out.println(Thread.currentThread().getName() + "任务执行结束");这一句话没有打印，因为CompletableFuture.supplyAsync异步线程是守护线程，主线程执行结束后，守护线程会立即结束

main执行开始

main执行结束

ForkJoinPool.commonPool-worker-1执行1

ForkJoinPool.commonPool-worker-1是否是守护线程-true

3.3.3 示例-传入参数pool

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {

        ForkJoinPool pool = new ForkJoinPool();

        System.out.println(Thread.currentThread().getName() + "执行开始");

        CompletableFuture.runAsync(()->{

            System.out.println(Thread.currentThread().getName() + "执行1");

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

            try {

                Thread.sleep(2000);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            System.out.println(Thread.currentThread().getName() + "任务执行结束");

        },pool);

        System.out.println(Thread.currentThread().getName() + "执行结束");

    }

执行结果

main执行开始

main执行结束

ForkJoinPool-1-worker-1执行1

ForkJoinPool-1-worker-1是否是守护线程-true

注意，这里runAsync传入了一个参数-线程池。supplyAsync和runAsync都可以传入一个线程池作为参数。不传时使用默认的线程池，传入了就使用传入的线程池

3.3.4 小结

　　1）supplyAsync 和 runAsync都会发起一个异步任务

　　2）supplyAsync 和 runAsync线程都是守护线程

　　3）supplyAsync 和 runAsync都可以传入一个线程池，传入了就使用传入的线程池

　　3）supplyAsync有返回值，runAsync没有返回值

3.4 异步回调方法 (thenApplyAsync/thenRunAsync/thenAcceptAsync)和(thenApply/thenRun/thenAccept)

　　thenApply 表示某个任务执行完成后执行的动作，即回调方法，会将该任务的执行结果即方法返回值作为入参传递到回调方法中

3.4.1 示例thenApplyAsync

接收任务执行完成返回的值做为参数，同时自己也有返回值

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {

        ForkJoinPool pool = new ForkJoinPool();

        System.out.println(Thread.currentThread().getName() + "执行开始");

        CompletableFuture.supplyAsync(()->{

            System.out.println(Thread.currentThread().getName() + "执行1");

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

            System.out.println(Thread.currentThread().getName() + "任务执行结束");

            return 111;

        },pool).thenApplyAsync((jobresult)->{

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

            try {

                Thread.sleep(2000);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            System.out.println("supplyAsync的任务完成了，传入执行结果回调thenApplyAsync");

            return jobresult*2;

        });

        System.out.println(Thread.currentThread().getName() + "执行结束");

    }

执行结果，发现thenApplyAsync回调中的线程和supplyAsync任务的线程不是从同一个线程池取的。因为supplyAsync传入了pool参数，使用的是我们创建的线程池。thenApplyAsync没有传入(也可以传入线程池参数)，使用的是默认的线程池

同时thenApplyAsync回调线程也是守护线程，主线程结束了，它就立即结束

main执行开始

ForkJoinPool-1-worker-1执行1

ForkJoinPool-1-worker-1是否是守护线程-true

ForkJoinPool-1-worker-1任务执行结束

main执行结束

ForkJoinPool.commonPool-worker-1是否是守护线程-true

3.4.2 示例-传入参数pool

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {

        ForkJoinPool pool = new ForkJoinPool();

        System.out.println(Thread.currentThread().getName() + "执行开始");

        CompletableFuture.supplyAsync(()->{

            System.out.println(Thread.currentThread().getName() + "执行1");

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

            System.out.println(Thread.currentThread().getName() + "任务执行结束");

            return 111;

        },pool).thenRunAsync(()->{

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

            try {

                Thread.sleep(2000);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            System.out.println("supplyAsync的任务完成了，传入执行结果回调thenApplyAsync");

        },pool);

        System.out.println(Thread.currentThread().getName() + "执行结束");

    }

执行结果，任务和回调使用的都是自己创建的线程池

main执行开始

ForkJoinPool-1-worker-1执行1

ForkJoinPool-1-worker-1是否是守护线程-true

ForkJoinPool-1-worker-1任务执行结束

main执行结束

ForkJoinPool-1-worker-1是否是守护线程-true

3.4.3 thenRunAsync

　　不用参数，也没有返回值

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {

        ForkJoinPool pool = new ForkJoinPool();

        System.out.println(Thread.currentThread().getName() + "执行开始");

        CompletableFuture.supplyAsync(()->{

            System.out.println(Thread.currentThread().getName() + "执行1");

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

            System.out.println(Thread.currentThread().getName() + "任务执行结束");

            return 111;

        },pool).thenRunAsync(()->{

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

            try {

                Thread.sleep(2000);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            System.out.println("supplyAsync的任务完成了，传入执行结果回调thenApplyAsync");

        });

        System.out.println(Thread.currentThread().getName() + "执行结束");

    }

执行结果，出了不接收参数无返回值外，其他的和thenApplyAsync一样

main执行开始

ForkJoinPool-1-worker-1执行1

ForkJoinPool-1-worker-1是否是守护线程-true

ForkJoinPool-1-worker-1任务执行结束

main执行结束

ForkJoinPool.commonPool-worker-1是否是守护线程-true

3.4.4 示例

不接受参数，有返回值thenAcceptAsync

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {

        ForkJoinPool pool = new ForkJoinPool();

        System.out.println(Thread.currentThread().getName() + "执行开始");

        CompletableFuture.supplyAsync(()->{

            System.out.println(Thread.currentThread().getName() + "执行1");

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

            System.out.println(Thread.currentThread().getName() + "任务执行结束");

            return 111;

        },pool).thenAcceptAsync((a)->{

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

            try {

                Thread.sleep(2000);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            System.out.println("supplyAsync的任务完成了，传入执行结果回调thenApplyAsync");

        });

        System.out.println(Thread.currentThread().getName() + "执行结束");

    }

执行结果

main执行开始

ForkJoinPool-1-worker-1执行1

ForkJoinPool-1-worker-1是否是守护线程-true

ForkJoinPool-1-worker-1任务执行结束

main执行结束

ForkJoinPool.commonPool-worker-1是否是守护线程-true

3.4.5 示例thenApply

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {

        ForkJoinPool pool = new ForkJoinPool();

        System.out.println(Thread.currentThread().getName() + "执行开始");

        CompletableFuture.supplyAsync(()->{

            System.out.println(Thread.currentThread().getName() + "执行1");

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

            System.out.println(Thread.currentThread().getName() + "任务执行结束");

            return 111;

        },pool).thenApply((a)->{

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

            try {

                Thread.sleep(2000);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            System.out.println("supplyAsync的任务完成了，传入执行结果回调thenApplyAsync");

            return a;

        });

        System.out.println(Thread.currentThread().getName() + "执行结束");

    }

执行结果，它不能传入pool作为参数。注意到，它里面用的线程是main主线程

main执行开始

ForkJoinPool-1-worker-1执行1

ForkJoinPool-1-worker-1是否是守护线程-true

ForkJoinPool-1-worker-1任务执行结束

main是否是守护线程-false

supplyAsync的任务完成了，传入执行结果回调thenApplyAsync

main执行结束

3.4.6 示例thenRun

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {

        ForkJoinPool pool = new ForkJoinPool();

        System.out.println(Thread.currentThread().getName() + "执行开始");

        CompletableFuture.supplyAsync(()->{

            System.out.println(Thread.currentThread().getName() + "执行1");

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

            System.out.println(Thread.currentThread().getName() + "任务执行结束");

            return 111;

        },pool).thenRun(()->{

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

            try {

                Thread.sleep(2000);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            System.out.println("supplyAsync的任务完成了，传入执行结果回调thenApplyAsync");

        });

        System.out.println(Thread.currentThread().getName() + "执行结束");

    }

执行结果，它不能传入pool作为参数。它里面用的线程是main主线程

main执行开始

ForkJoinPool-1-worker-1执行1

ForkJoinPool-1-worker-1是否是守护线程-true

ForkJoinPool-1-worker-1任务执行结束

main是否是守护线程-false

supplyAsync的任务完成了，传入执行结果回调thenApplyAsync

main执行结束

3.4.7 示例thenAccept

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {

        ForkJoinPool pool = new ForkJoinPool();

        System.out.println(Thread.currentThread().getName() + "执行开始");

        CompletableFuture.supplyAsync(()->{

            System.out.println(Thread.currentThread().getName() + "执行1");

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

            System.out.println(Thread.currentThread().getName() + "任务执行结束");

            return 111;

        },pool).thenAccept((a)->{

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

            try {

                Thread.sleep(2000);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            System.out.println("supplyAsync的任务完成了，传入执行结果回调thenApplyAsync");

        });

        System.out.println(Thread.currentThread().getName() + "执行结束");

    }

执行结果，它不能传入pool作为参数。它里面用的线程是main主线程

main执行开始

ForkJoinPool-1-worker-1执行1

ForkJoinPool-1-worker-1是否是守护线程-true

ForkJoinPool-1-worker-1任务执行结束

main是否是守护线程-false

supplyAsync的任务完成了，传入执行结果回调thenApplyAsync

main执行结束

3.4.8 小结

　　1）六个方法都是任务执行完成后的回调方法

　　2）thenApplyAsync和 thenRunAsync和thenAcceptAsync

　　　　都可以传入一个线程池，传入了就使用传入的线程池，不传入使用默认的线程池

　　　　线程都是守护线程，在主线程结束后，会结束。

　　　　thenApplyAsync接收一个参数，有一个返回值。thenRunAsync没有参数，也没有返回值，thenAcceptAsync有一个参数，没有返回值

　　3）thenApply和 thenRun和thenAccept(和前三个的区别)

　　　　没有pool作为参数，使用的是还未回收的线程(有可能是上个任务的线程，有可能是主线程)

3.5 whenComplete方法

　　whenComplete是当某个任务执行完成后执行的回调方法，会将执行结果或者执行期间抛出的异常传递给回调方法

3.5.1 示例1(whenComplete)

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {

        System.out.println(Thread.currentThread().getName() + "执行开始");

          CompletableFuture.supplyAsync(()->{

            System.out.println(Thread.currentThread().getName() + "执行1");

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

            try {

                Thread.sleep(2000);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            return 1;

        }).whenComplete((v,e)->{ //第一个参数是返回值 第二个参数是异常

            System.out.println(Thread.currentThread().getName() + "执行2");

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

                if(e == null){

                    System.out.println(v);

                }

        }).exceptionally((e)->{

            e.printStackTrace();

            return 0;

        });

        //Thread.sleep(3000);

        System.out.println(Thread.currentThread().getName() + "执行结束");

    }

执行结果，发现whenComplete里面的没有执行，因为主线程执行结束了，守护线程也就结束了，CompletableFuture创建的是守护线程

　　 main执行开始

    ForkJoinPool.commonPool-worker-1执行1

    ForkJoinPool.commonPool-worker-1是否是守护线程-true

    main执行结束

把//Thread.sleep(3000);的注释解开，执行结果，whenComplete里面的执行了，因为主线程等待了3秒钟

　　main执行开始

    ForkJoinPool.commonPool-worker-1执行1

    ForkJoinPool.commonPool-worker-1是否是守护线程-true

    ForkJoinPool.commonPool-worker-1执行2

    ForkJoinPool.commonPool-worker-1是否是守护线程-true

    1

    main执行结束

3.5.2 示例2(whenComplete+get)

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {

        System.out.println(Thread.currentThread().getName() + "执行开始");

        CompletableFuture.supplyAsync(()->{

            System.out.println(Thread.currentThread().getName() + "执行1");

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

            try {

                Thread.sleep(2000);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            return 1;

        }).whenComplete((v,e)->{

            System.out.println(Thread.currentThread().getName() + "执行2");

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

                if(e == null){

                    System.out.println(v);

                }

        }).exceptionally((e)->{

            e.printStackTrace();

            return 0;

        }).get();

        System.out.println(Thread.currentThread().getName() + "执行结束");

    }

执行结果，whenComplete正常执行，因为get会导致主线程阻塞

　　main执行开始

    ForkJoinPool.commonPool-worker-1执行1

    ForkJoinPool.commonPool-worker-1是否是守护线程-true

    ForkJoinPool.commonPool-worker-1执行2

    ForkJoinPool.commonPool-worker-1是否是守护线程-true

            1

    main执行结束

3.5.3 示例3(whenComplete+get(超时时间))

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {

        System.out.println(Thread.currentThread().getName() + "执行开始");

        CompletableFuture.supplyAsync(()->{

            System.out.println(Thread.currentThread().getName() + "执行1");

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

            try {

                Thread.sleep(2000);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            return 1;

        }).whenComplete((v,e)->{

            System.out.println(Thread.currentThread().getName() + "执行2");

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

                if(e == null){

                    System.out.println(v);

                }

        }).exceptionally((e)->{

            e.printStackTrace();

            return 0;

        }).get(1, TimeUnit.SECONDS);

        System.out.println(Thread.currentThread().getName() + "执行结束");

    }

执行结果，超时报错，结束

　　main执行开始

　　ForkJoinPool.commonPool-worker-1执行1

　　ForkJoinPool.commonPool-worker-1是否是守护线程-true

　　Exception in thread "main" java.util.concurrent.TimeoutException

    at java.util.concurrent.CompletableFuture.timedGet(CompletableFuture.java:1771)

    at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1915)

    at com.ruoyi.weixin.user.MyTest.JucTest.CompletableFutureTest2.main(CompletableFutureTest2.java:40)

3.5.4 示例4(whenComplete+join(超时时间))

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {

        System.out.println(Thread.currentThread().getName() + "执行开始");

        CompletableFuture.supplyAsync(()->{

            System.out.println(Thread.currentThread().getName() + "执行1");

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

            try {

                Thread.sleep(2000);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            return 1;

        }).whenComplete((v,e)->{

            System.out.println(Thread.currentThread().getName() + "执行2");

            System.out.println( Thread.currentThread().getName() + "是否是守护线程-" + Thread.currentThread().isDaemon());

                if(e == null){

                    System.out.println(v);

                }

        }).exceptionally((e)->{

            e.printStackTrace();

            return 0;

        }).join();

        System.out.println(Thread.currentThread().getName() + "执行结束");

    }

执行结果，join和get方法一样，都是去获取结果，都会造成阻塞

main执行开始

ForkJoinPool.commonPool-worker-1执行1

ForkJoinPool.commonPool-worker-1是否是守护线程-true

ForkJoinPool.commonPool-worker-1执行2

ForkJoinPool.commonPool-worker-1是否是守护线程-true

1

main执行结束

3.5.5 小结

　　1）whenComplete是任务执行完成后的回调

　　2）whenComplete是守护线程

3）whenComplete不会阻塞主线程

4）get方法会阻塞主线程

　　5）get(等待时间)，超时会直接报错

3.6 CompletableFuture的优势示例

@Data

@AllArgsConstructor

@NoArgsConstructor

@Accessors(chain = true)

public class Book {

    private String name;

    private Double price;

}

public class CompletableFutureTest6 {

    public static void main(String[] args) {

        List<String> li = new ArrayList<>();

        li.add("京东");

        li.add("淘宝");

        li.add("当当");

        li.add("唯品会");

        li.add("拼多多");

        if(true){//一个接一个去拿

            long l1 = System.currentTimeMillis();

            li.stream().map(shop -> String.format("%s 的价格是:%f",shop ,getBook(shop,"问道").getPrice())).

                    collect(Collectors.toList()).

                    stream().

                    forEach(System.out::println);;

            System.out.println(System.currentTimeMillis()-l1);

        }

        if(true){//异步多个同时去拿

            long l1 = System.currentTimeMillis();

            li.stream().

                    map(shop-> CompletableFuture.supplyAsync(()->String.format("%s 的价格是:%f",shop ,getBook(shop,"问道").getPrice()))).

                    collect(Collectors.toList()).

                    stream().

                    map(CompletableFuture::join).forEach(System.out::println);

            System.out.println(System.currentTimeMillis()-l1);

        }

    }

    public static Book getBook(String shopname,String bookname){

        System.out.println(Thread.currentThread().getName() + ":到" + shopname + "取书-" + bookname);

        try {

            Thread.sleep(1000);

        } catch (InterruptedException e) {

            e.printStackTrace();

        }

        Book b = new Book(bookname, ThreadLocalRandom.current().nextDouble());

        return b;

    }

}

执行结果，发现使用CompletableFuture异步获取大大缩短了执行时间。因为它是多个线程同时去获取的。

main:到京东取书-问道

main:到淘宝取书-问道

main:到当当取书-问道

main:到唯品会取书-问道

main:到拼多多取书-问道

京东 的价格是:0.316333

淘宝 的价格是:0.593280

当当 的价格是:0.761204

唯品会 的价格是:0.677869

拼多多 的价格是:0.686680

5070

ForkJoinPool.commonPool-worker-1:到京东取书-问道

ForkJoinPool.commonPool-worker-2:到淘宝取书-问道

ForkJoinPool.commonPool-worker-3:到当当取书-问道

ForkJoinPool.commonPool-worker-4:到唯品会取书-问道

ForkJoinPool.commonPool-worker-5:到拼多多取书-问道

京东 的价格是:0.507088

淘宝 的价格是:0.402202

当当 的价格是:0.537196

唯品会 的价格是:0.261127

拼多多 的价格是:0.293971

1006

3.7 CompletableFuture常用方法简介

3.7.1 获取结果

　　1）public T get()，获取结果，会造成阻塞

　　2）public T get(long timeout, TimeUnit unit)，获取结果，超时报错

　　3）public T getNow(T valueIfAbsent)，立即获取结果不会造成阻塞。调用该方法时，如果任务已经执行完了，返回任务的结果，如果任务没有执行完，获取到的是预设的值valueIfAbsent

　　4）public T join()，获取结果，会造成阻塞

3.7.2 主动结束任务

　　1）public boolean complete(T value) 主动去结束任务。调用本方法时，如果任务已经结束，返回false。如果任务没有结束，立即结束任务，返回true。且把参数value作为任务的返回值

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {

        CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> {

            try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); }

            return 111;

        });

        Thread.sleep(2100);

        System.out.println(integerCompletableFuture.complete(222) + "|" + integerCompletableFuture.get());

    }

执行结果。执行complete方法时，任务已结束，返回的是false，get获取到的是任务的返回值111

false|111

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {

        CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> {

            try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); }

            return 111;

        });

        System.out.println(integerCompletableFuture.complete(222) + "|" + integerCompletableFuture.get());

    }

执行结果。调用complete方法时，任务还在执行中，立即结束任务，返回为true。get获取到的是complete传入的预设值222

true|222

3.7.3 对计算结果进行处理

3.7.3.1　thenApplyAsync和 thenRunAsync和thenAcceptAsync和thenApply和 thenRun和thenAccept(再回顾下这六个方法)

　　1）六个方法都是任务执行完成后的回调方法

　　2）thenApplyAsync和 thenRunAsync和thenAcceptAsync

　　　　都可以传入一个线程池，传入了就使用传入的线程池，不传入使用默认的线程池

　　　　线程都是守护线程，在主线程结束后，会结束。

　　　　thenApplyAsync接收一个参数，有一个返回值。thenRunAsync没有参数，也没有返回值，thenAcceptAsync有一个参数，没有返回值

　　3）thenApply和 thenRun和thenAccept(和前三个的区别)

　　　　没有pool作为参数，使用的是还未回收的线程(有可能是上个任务的线程，有可能是主线程)

　　带上Async和没有Async的区别就是，带上Async交给线程池管理，不带Async使用的是还未回收的线程(有可能是上个任务的线程，有可能是主线程)

3.7.3.2 handle

　　和thenApply差不多，不过它执行过程中出现异常，还可以继续执行下去

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {

        CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> {

            return 111;

        }).thenApply((v)->{

            System.out.println("第一个thenApply：" + v);

            int i = 1/0;

            return v * 2;

        }).thenApply((v)->{

            System.out.println("第二个thenApply：" + v);

            return v * 3;

        }).exceptionally(e->{

            e.printStackTrace();

            return -1;

        });

    }

执行结果，第一个thenApply出错后，第二个thenApply没有执行，直接来到exceptionally异常处理

第一个thenApply：111

java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero

    at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273)

    at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280)

    at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:604)

    at java.util.concurrent.CompletableFuture.uniApplyStage(CompletableFuture.java:614)

    at java.util.concurrent.CompletableFuture.thenApply(CompletableFuture.java:1983)

    at com.ruoyi.weixin.user.MyTest.JucTest.CompletableFutureTest19.main(CompletableFutureTest19.java:20)

Caused by: java.lang.ArithmeticException: / by zero

    at com.ruoyi.weixin.user.MyTest.JucTest.CompletableFutureTest19.lambda$main$1(CompletableFutureTest19.java:22)

    at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)

    ... 3 more

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {

        CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> {

            return 111;

        }).handle((v,e)->{

            System.out.println("第一个handle：" + v);

            int i = 1/0;

            return v * 2;

        }).handle((v,e)->{

            System.out.println("第二个handle：" + v);

            return v * 3;

        }).exceptionally(e->{

            e.printStackTrace();

            return -1;

        });

    }

执行结果，第一个handle出错后，返回null值，第二个handle继续执行，最后才回到exceptionally异常梳理

第一个handle：111

第二个handle：null

java.util.concurrent.CompletionException: java.lang.NullPointerException

    at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273)

    at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280)

    at java.util.concurrent.CompletableFuture.uniHandle(CompletableFuture.java:824)

    at java.util.concurrent.CompletableFuture.uniHandleStage(CompletableFuture.java:834)

    at java.util.concurrent.CompletableFuture.handle(CompletableFuture.java:2155)

    at com.ruoyi.weixin.user.MyTest.JucTest.CompletableFutureTest18.main(CompletableFutureTest18.java:24)

Caused by: java.lang.NullPointerException

    at com.ruoyi.weixin.user.MyTest.JucTest.CompletableFutureTest18.lambda$main$2(CompletableFutureTest18.java:26)

    at java.util.concurrent.CompletableFuture.uniHandle(CompletableFuture.java:822)

    ... 3 more

3.7.3.3 thenCompose

　　thenCompose允许将两个异步操作进行流水线，第一个操作完成时,将其结果作为参数传递给第二个操作

<U> CompletionStage<U> thenApply(Function<? super T,? extends U> fn)

<U> CompletionStage<U> thenCompose(Function<? super T,? extends CompletionStage<U>> fn)

可以看到，两个方法的返回值都是CompletionStage<U>，不同之处在于它们的传入参数fn.

对于thenApply，fn函数是一个对一个已完成的stage或者说CompletableFuture的的返回值进行计算、操作；
对于thenCompose，fn函数是对另一个CompletableFuture进行计算、操作。

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {

        CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {

            System.out.println("任务1执行");

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

            System.out.println("完成任务1执行");

            return 100;

        }).thenApply(num -> {

            System.out.println("任务2执行");

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

            System.out.println("完成任务2执行");

            return num + " to String";

        });

        CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> {

            System.out.println("任务3执行");

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

            System.out.println("完成任务3执行");

            return 100;

        }).thenCompose(num -> CompletableFuture.supplyAsync(() -> {

            System.out.println("任务4执行");

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

            System.out.println("完成任务4执行");

            return num + " to String";

        }));

        System.out.println(f1.join());

        System.out.println(f2.join());

    }

执行结果

　　thenApply是任务1和任务2，任务2是在任务1完成后再执行的

　　thenCompose的是任务3和任务4，任务4是在任务3完成后再执行的

　　也就是说thenApply和thenCompose的任务都是顺序执行，后一个任务依赖前一个任务

任务1执行

任务3执行

完成任务1执行

任务2执行

完成任务3执行

任务4执行

完成任务2执行

100 to String

完成任务4执行

100 to String

Process finished with exit code 0

3.7.4 竞争上岗applyToEither

　　获取最先执行完的任务的结果，然后采取这个任务及之后任务的结果处理函数进行处理

static void completableFutureTe3() throws ExecutionException, InterruptedException {

        String re = CompletableFuture.supplyAsync(() -> {

            try { Thread.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); }

            System.out.println("执行1号任务");

            return "1号任务返回值";

        }).applyToEither(CompletableFuture.supplyAsync(() -> {

            try { Thread.sleep(200); } catch (InterruptedException e) { e.printStackTrace(); }

            System.out.println("执行2号任务");

            return "  2号任务返回值  ";

        }), x -> x + "*对结果第一次处理*").applyToEither(CompletableFuture.supplyAsync(() -> {

            try { Thread.sleep(200); } catch (InterruptedException e) { e.printStackTrace(); }

            System.out.println("执行3号任务");

            return "2号任务返回值";

        }), y -> y + "*对结果第二次处理*").get();

        System.out.println(re);

    }

执行结果

　　任务1、2、3共三个任务同时执行，谁先完成就取谁的结果。任务一所需时间最短，所以取得任务一的返回值，然后再经过了两次处理

执行1号任务
1号任务返回值*对结果第一次处理**对结果第二次处理*

　　把睡眠时间调整下，让第二个任务先执行完

 //几个任务同时执行，获取最先执行完的结果，再进行处理

    static void completableFutureTe3() throws ExecutionException, InterruptedException {

        String re = CompletableFuture.supplyAsync(() -> {

            try { Thread.sleep(200); } catch (InterruptedException e) { e.printStackTrace(); }

            System.out.println("执行1号任务");

            return "1号任务返回值";

        }).applyToEither(CompletableFuture.supplyAsync(() -> {

            try { Thread.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); }

            System.out.println("执行2号任务");

            return "  2号任务返回值  ";

        }), x -> x + "*对结果第一次处理*").applyToEither(CompletableFuture.supplyAsync(() -> {

            try { Thread.sleep(200); } catch (InterruptedException e) { e.printStackTrace(); }

            System.out.println("执行3号任务");

            return "3号任务返回值";

        }), y -> y + "*对结果第二次处理*").get();

        System.out.println(re);

    }

执行结果

执行2号任务

  2号任务返回值  *对结果第一次处理**对结果第二次处理*

把睡眠时间调整下，让第三个任务先执行完

static void completableFutureTe3() throws ExecutionException, InterruptedException {

        String re = CompletableFuture.supplyAsync(() -> {

            try { Thread.sleep(200); } catch (InterruptedException e) { e.printStackTrace(); }

            System.out.println("执行1号任务");

            return "1号任务返回值";

        }).applyToEither(CompletableFuture.supplyAsync(() -> {

            try { Thread.sleep(200); } catch (InterruptedException e) { e.printStackTrace(); }

            System.out.println("执行2号任务");

            return "  2号任务返回值  ";

        }), x -> x + "*对结果第一次处理*").applyToEither(CompletableFuture.supplyAsync(() -> {

            try { Thread.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); }

            System.out.println("执行3号任务");

            return "3号任务返回值";

        }), y -> y + "*对结果第二次处理*").get();

        System.out.println(re);

    }

执行结果

执行3号任务
3号任务返回值*对结果第二次处理*

注意，此时和上面两次有所不同，上面对结果都进行了两次处理，而这里只进行了一次。说明取得某个任务的结果后，只会采取这个任务及之后的结果处理函数对结果进行处理。因为这里取得的是第三个任务的结果，所以只采用了这个任务的结果处理函数进行处理

3.7.4 对计算结果进行合并thenCombine

3.7.4.1 示例

    public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {

        Integer join = CompletableFuture.supplyAsync(() -> {

            System.out.println("执行第一个任务");

            try {

                Thread.sleep(1000);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            System.out.println("完成执行第一个任务");

            return 10;

        }).thenCombine(CompletableFuture.supplyAsync(() -> {

            System.out.println("执行第二个任务");

            try {

                Thread.sleep(2000);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            System.out.println("完成执行第二个任务");

            return 20;

        }), (a, b) -> {

            return a + b;

        }).thenCombine(CompletableFuture.supplyAsync(() -> {

            System.out.println("执行第三个任务");

            try {

                Thread.sleep(1500);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            System.out.println("完成执行第三个任务");

            return 30;

        }), (a, b) -> {

            return a + b;

        }).join();

        System.out.println(join);

    }

}

执行结果

　　共三个任务同时执行，先把任务1和2的结果进行处理，再把处理后的结果和第三个任务的结果进行处理。

执行第一个任务

执行第二个任务

执行第三个任务

完成执行第一个任务

完成执行第三个任务

完成执行第二个任务

60

巴特西

线程基础知识02-CompletableFuture

1 简介

2 Future

2.1 源码

2.2 示例2

2.3 示例2(不见不散)

2.4 示例3(过时不候)

2.5 示例4

2.6 小结

3 CompletableFuture

3.1 简介

3.2 继承关系图

3.3 异步任务方法supplyAsync / runAsync

3.4 异步回调方法 (thenApplyAsync/thenRunAsync/thenAcceptAsync)和(thenApply/thenRun/thenAccept)

3.5 whenComplete方法

3.6 CompletableFuture的优势示例

3.7 CompletableFuture常用方法简介

最新文章

热门文章