CompletableFuture

2022-11-20 | 0 评论 | 0 浏览

CompletableFuture

Future和Callable接口

Future接口(FutureTask实现类)定义了操作异步任务执行一些方法，如获取异步任务的执行结果、取消任务的执行、判断任务是否被取消、判断任务执行是否完毕等。（异步：可以被叫停，可以被取消）
一句话：Future接口可以为主线程开一个分支任务，专门为主线程处理耗时和费力的复杂业务。

eg.比如主线程让一个子线程去执行任务，子线程可能比较耗时，启动子线程开始执行任务后，主线程就去做其他事情了，过了一会才去获取子任务的执行结果。老师在上课，但是口渴，于是让班长这个线程去买水，自己可以继续上课，实现了异步任务。
有个目的：异步多线程任务执行且有返回结果，三个特点：多线程/有返回/异步任务（班长作为老师去买水作为新启动的异步多线程任务且买到水有结果返回）

FutureTask实现类

FutureTak(实现了RunnableFuture接口，RunnableFuture接口又继承了Runnable和Future接口)
- 在源码可以看到，他既继承了RunnableFuture接口，也在构造方法中实现了Callable接口（有返回值、可抛出异常）和Runnable接口

(ctrl+alt+u)

FutureTask有两个构造方法：

FutureTask(Callable<V> callable)

FutureTask(Runnable runnable, V result)

callable和runnable创建区别在于callable有返回值且call方法抛出异常，而runnable没有返回值不抛出异常

完成上面目的的代码 - 多线程/有返回/异步
- 一个主线程，一个mythread异步执行并返回了"hello callable"

public class CompletableFutureDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //通过Callable构造方法创建FutureTask
        FutureTask<String> futureTask = new FutureTask<>(new MyThread());
        //创建线程
        Thread t1 = new Thread(futureTask, "t1");
        t1.start();
        //获取线程返回值
        String result = futureTask.get();
        System.out.println(result);
    }
}

class MyThread implements Callable<String> {
    @Override
    public String call() throws Exception {
        System.out.println("-----come in call() ");
        return "hello Callable";
    }
}
//结果
//-----come in call() ----异步执行
//hello Callable 返回值

Future到CompletableFuture

Future优点

future+线程池异步多线程任务配合，能显著提高程序的执行效率。
方案一，3个任务1个main线程处理，大概1130ms

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

        //3个任务，目前只有一个线程main来处理，请问耗时多少？

        long startTime = System.currentTimeMillis();
        //暂停毫秒
        try { TimeUnit.MILLISECONDS.sleep(500); } catch (InterruptedException e) { e.printStackTrace(); }
        try { TimeUnit.MILLISECONDS.sleep(300); } catch (InterruptedException e) { e.printStackTrace(); }
        try { TimeUnit.MILLISECONDS.sleep(300); } catch (InterruptedException e) { e.printStackTrace(); }

        long endTime = System.currentTimeMillis();
        System.out.println("----costTime: "+(endTime - startTime) +" 毫秒");

        System.out.println(Thread.currentThread().getName()+"\t -----end");
        }
}
//结果
//----costTime: 1167 毫秒
//main	 -----end

方案二，3个任务3个线程，利用线程池（假如每次new一个Thread，太浪费资源，会有GC这些工作），大概400毫秒，futureTask.get()方法会阻塞主线程执行时间会更长一些。

public class FutureThreadPoolDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //3个任务，目前开启多个异步任务线程来处理，请问耗时多少？
        //创建线程池
        ExecutorService threadPool = Executors.newFixedThreadPool(3);

        long startTime = System.currentTimeMillis();

        FutureTask<String> futureTask1 = new FutureTask<String>(() -> {
            try { TimeUnit.MILLISECONDS.sleep(500); } catch (InterruptedException e) { e.printStackTrace(); }
            return "task1 over";
        });
        threadPool.submit(futureTask1);

        FutureTask<String> futureTask2 = new FutureTask<String>(() -> {
            try { TimeUnit.MILLISECONDS.sleep(300); } catch (InterruptedException e) { e.printStackTrace(); }
            return "task2 over";
        });
        threadPool.submit(futureTask2);

        //调用futureTask get方法会阻塞主线程，获取到返回值后继续执行主线程
        System.out.println(futureTask1.get());
        System.out.println(futureTask2.get());

        try { TimeUnit.MILLISECONDS.sleep(300); } catch (InterruptedException e) { e.printStackTrace(); }

        long endTime = System.currentTimeMillis();
        System.out.println("----costTime: "+(endTime - startTime) +" 毫秒");


        System.out.println(Thread.currentThread().getName()+"\t -----end");
        threadPool.shutdown();


    }
//结果
//task1 over
//task2 over
//----costTime: 862 毫秒
//main	 -----end

Future缺点

1 get()阻塞

一旦调用get()方法，不管是否计算完成，都会导致阻塞（所以一般get方法放到最后）

public class FutureAPIDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
        FutureTask<String> futureTask = new FutureTask<String>(() -> {
            System.out.println(Thread.currentThread().getName() + "\t -----come in");
            try {
                TimeUnit.SECONDS.sleep(5);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "task over";
        });
        Thread t1 = new Thread(futureTask, "t1");
        t1.start();

        System.out.println(Thread.currentThread().getName() + "\t ----忙其它任务了");
        // get容易导致阻塞，一般建议放在程序后面，一旦调用不见不散，非要等到结果才会离开，不管你是否计算完成，容易程序堵塞。
        System.out.println(futureTask.get());
    }
}
//结果
//main	 ----忙其它任务了
//t1	 -----come in
//task over

如果我们将System.out.println(futureTask.get());和System.out.println(Thread.currentThread().getName() + "\t ----忙其它任务了");调换位置

get方法会阻塞主线程，最终输出结果如下:

t1	 -----come in
task over
main	 ----忙其它任务了

public class FutureAPIDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
        FutureTask<String> futureTask = new FutureTask<String>(() -> {
            System.out.println(Thread.currentThread().getName() + "\t -----come in");
            try {
                TimeUnit.SECONDS.sleep(5);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "task over";
        });
        Thread t1 = new Thread(futureTask, "t1");
        t1.start();
        // 假如我不愿意等待很长时间，我希望过时不候，可以自动离开.
        // 获取子线程返回结果，等待3秒如果没有返回就抛出异常
        System.out.println(futureTask.get(3,TimeUnit.SECONDS));
        System.out.println(Thread.currentThread().getName() + "\t ----忙其它任务了");

    }
}

2 isDone()轮询

利用if(futureTask.isDone())的方式，可以在子线程在结束之后在调用get(),但需要轮询判断，所以会造成cpu的消耗

public class FutureAPIDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
        FutureTask<String> futureTask = new FutureTask<String>(() -> {
            System.out.println(Thread.currentThread().getName() + "\t -----come in");
            try {
                TimeUnit.SECONDS.sleep(5);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "task over";
        });
        Thread t1 = new Thread(futureTask, "t1");
        t1.start();

        System.out.println(Thread.currentThread().getName() + "\t ----忙其它任务了");
        //get容易导致阻塞，一般建议放在程序后面，一旦调用不见不散，非要等到结果才会离开，不管你是否计算完成，容易程序堵塞。
        //System.out.println(futureTask.get());
        //假如我不愿意等待很长时间，我希望过时不候，可以自动离开. 等待3秒如果没有获取到返回值
        //System.out.println(futureTask.get(3,TimeUnit.SECONDS));

        //轮询判断futureTask子线程是否执行完毕，执行完毕调用get方法获取返回值结果
        while (true) {
            if (futureTask.isDone()) {
                //执行完毕调用get方法获取返回值结果
                System.out.println(futureTask.get());
                break;
            } else {
                //暂停毫秒
                try {
                    TimeUnit.MILLISECONDS.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("正在处理中，不要再催了，越催越慢 ，再催熄火");
            }
        }
    }
}
//结果
main	 ----忙其它任务了
t1	 -----come in
正在处理中，不要再催了，越催越慢 ，再催熄火
正在处理中，不要再催了，越催越慢 ，再催熄火
正在处理中，不要再催了，越催越慢 ，再催熄火
正在处理中，不要再催了，越催越慢 ，再催熄火
正在处理中，不要再催了，越催越慢 ，再催熄火
正在处理中，不要再催了，越催越慢 ，再催熄火
正在处理中，不要再催了，越催越慢 ，再催熄火
正在处理中，不要再催了，越催越慢 ，再催熄火
正在处理中，不要再催了，越催越慢 ，再催熄火
正在处理中，不要再催了，越催越慢 ，再催熄火
task over

Future应用现状

对于简单的业务场景使用Future完全OK
- 回调通知
  - 前面的isDone()方法耗费cpu资源，一般应该还是利用回调函数，在Future结束时自动调用该回调函数。应对Future的完成时间，完成了可以告诉我，也就是我们的回调通知
- 创建异步任务
  - Future+线程池配合
    多个任务前后依赖可以组合处理（水煮鱼）
  - 想将多个异步任务的计算结果组合起来，后一个异步任务的计算结果需要前一个异步任务的值，将两个或多个异步计算合成一个异步计算，这几个异步计算相互独立，同时后面这个又依赖前一个处理的结果
  - 比如买鱼-加料-烹饪
- 对计算速度选最快完成的（并返回结果）
- 当Future集合中某个任务最快结束时，返回结果，返回第一名处理结果。
- …

Future应用于一些复杂的业务场景就会暴露出很多弊端和不足

CompletableFuture基本介绍

阻塞的方式和异步编程的设计理念相违背，而轮询的方式会消耗无畏的CPU资源。因此，JDK8设计出CompletableFuture

CompletableFuture

public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {

在Java 8中， CompletableFuture提供了非常强大的Future的扩展功能，可以帮助我们简化异步编程的复杂性，并且提供了函数式编程的能力，可以通过回调的方式处理计算结果，也提供了转换和组合CompletableFuture的方法。
它可能代表一个明确完成的Future，也有可能代表一个完成阶段(Completion Stage) ，它支持在计算完成以后触发一些函数或执行某些动作。
它实现了Future和Completion Stage接口

CompletionStage

CompletionStage代表异步计算过程中的某一个阶段，一个阶段完成以后可能会触发另外一个阶段
一个阶段的计算执行可以是一个Function， Consumer或者Runnable。比如：

stage.then Apply(x->square(x) ) .then Accept(x->System.out.print(x) ) .then Run() ->System.out.print In())
，一个阶段的执行可能是被单个阶段的完成触发，也可能是由多个阶段一起触发。

核心的四个静态方法（分为两组）

利用核心的四个静态方法创建一个异步操作 | 不建议用new
关键就是 |有没有返回值|是否用了线程池|
参数说明：
- 没有指定Executor的方法，直接使用默认的ForkJoinPool.commPool()作为它的线程池执行异步代码。默认的ForkJoinPool.commPool()开启的线程是守护线程主线从结束之后会终止执行
- 如果指定线程池，则使用我们定义的或者特别指定的线程池执行异步代码。

runAsync无返回值

1 runAsync

public static CompletableFuture<Void> runAsync(Runnable runnable) {
        return asyncRunStage(asyncPool, runnable);
    }

public class CompletableFutureBuildDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<Void> voidCompletableFuture = CompletableFuture.runAsync(() -> {
            System.out.println(Thread.currentThread().getName());
            //停顿几秒线程
            try {
                TimeUnit.SECONDS.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        //CompletableFuture的get方法还是会阻塞主线程的执行
        System.out.println(voidCompletableFuture.get());
    }
}
//ForkJoinPool.commonPool-worker-9 //默认的线程池
//null --- 没有返回值

2 runAsync+线程池

public static CompletableFuture<Void> runAsync(Runnable runnable,
                                               Executor executor) {
    return asyncRunStage(screenExecutor(executor), runnable);
}

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

        ExecutorService executorService = Executors.newFixedThreadPool(3);//加入线程池

        CompletableFuture<Void> voidCompletableFuture = CompletableFuture.runAsync(() -> {

            System.out.println(Thread.currentThread().getName());
            //停顿几秒线程
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },executorService);
        System.out.println(voidCompletableFuture.get());
    }
}
//pool-1-thread-1   ----指定的线程池
//null ----没有返回值

supplyAsync有返回值

3 supplyAsync有返回值

public class CompletableFutureBuildDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //异步执行并接收返回值
        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName());
            //暂停几秒钟线程
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "hello supplyAsync";
        });
        //CompletableFuture的get方法还是会阻塞主线程的执行
        System.out.println(completableFuture.get());
        System.out.println(Thread.currentThread().getName());

    }
}
//结果
ForkJoinPool.commonPool-worker-9
hello supplyAsync
main
//没有指定线程池，则使用ForkJoinPool默认的线程池，需要注意的是ForkJoinPool创建的线程是守护线程，用户线程结束会停止运行，比如这里去掉get方法不会阻塞主线程的执行，最终输出结果为
main
ForkJoinPool.commonPool-worker-9

4 supplyAsync+线程池

public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,
                                                   Executor executor) {
    return asyncSupplyStage(screenExecutor(executor), supplier);
}

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

        ExecutorService executorService = Executors.newFixedThreadPool(3);//加入线程池

        CompletableFuture<String> objectCompletableFuture = CompletableFuture.supplyAsync(()->{
            System.out.println(Thread.currentThread().getName());
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "helllo supplyasync";
        },executorService);
        System.out.println(objectCompletableFuture.get());
    }
}
//pool-1-thread-1
//helllo supplyasync

CompletableFuture使用演示（日常使用）

基本功能

CompletableFuture可以完成Future的功能

public class CompletableFutureUseDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<Object> objectCompletableFuture = CompletableFuture.supplyAsync(()->{
            System.out.println(Thread.currentThread().getName()+"----副线程come in");
            int result = ThreadLocalRandom.current().nextInt(10);//产生一个随机数
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("1秒钟后出结果"+result);
            return result;
        });

        System.out.println(Thread.currentThread().getName()+"线程先去忙其他任务");
        System.out.println(objectCompletableFuture.get());
    }
}
//main线程先去忙其他任务
//ForkJoinPool.commonPool-worker-9----副线程come in
//1秒钟后出结果6
//6

减少阻塞和轮询whenComplete

CompletableFuture通过whenComplete来减少阻塞和轮询，子线程执行玩之后会回调whenComplete方法

public class CompletableFutureUseDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture.supplyAsync(()->{
            System.out.println(Thread.currentThread().getName()+"--------副线程come in");
            int result = ThreadLocalRandom.current().nextInt(10);//产生随机数
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return result;
        }).whenComplete((v,e) -> {//没有异常,v是result值，e是异常
            if(e == null){
                System.out.println("------------------计算完成，更新系统updateValue"+v);
            }
        }).exceptionally(e->{//有异常的情况
            e.printStackTrace();
            System.out.println("异常情况"+e.getCause()+"\t"+e.getMessage());
            return null;
        });

        //线程不要立刻结束，否则CompletableFuture默认使用的线程池会立刻关闭：暂停3秒钟线程
        //也可以使用自定义线程池，手动释放来解决
        System.out.println(Thread.currentThread().getName()+"线程先去忙其他任务");
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
//ForkJoinPool.commonPool-worker-9--------副线程come in（这里用的是默认的ForkJoinPool）
//main线程先去忙其他任务
//------------------计算完成，更新系统updateValue3

换用自定义线程池

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

        ExecutorService threadPool = Executors.newFixedThreadPool(3);

        try {
            CompletableFuture.supplyAsync(() -> {
                System.out.println(Thread.currentThread().getName() + "----come in");
                int result = ThreadLocalRandom.current().nextInt(10);
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("-----1秒钟后出结果：" + result);
                if (result > 2) {
                    int i = 10 / 0;
                }
                return result;
                //不论执行成功或者失败都会回调whenComplete方法
            }, threadPool).whenComplete((v, e) -> {
                if (e == null) {
                    System.out.println("-----计算完成，更新系统UpdateValue：" + v);
                }
                //执行异常调用exceptionally方法
            }).exceptionally(e -> {
                e.printStackTrace();
                System.out.println("异常情况：" + e.getCause() + "\t" + e.getMessage());
                return null;
            });

            System.out.println(Thread.currentThread().getName() + "线程先去忙其它任务");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //最后手动释放线程，自定的线程池创建的线程也是用户线程，这样就可以主线程执行完不必阻塞了
            threadPool.shutdown();
        }
    }
//正常执行的情况    
pool-1-thread-1----come in
main线程先去忙其它任务
-----1秒钟后出结果：2
-----计算完成，更新系统UpdateValue：2

CompletableFuture优点总结

异步任务结束时，会自动回调某个对象的方法；
主线程设置好毁掉后，不再关心异步任务的执行，异步任务之间可以顺序执行
异步任务出错时，会自动回调某个对象的方法。

CompletableFuture案例精讲

编程必备技能准备

函数式接口

函数式接口的定义：
- 任何接口，如果只包含唯一一个抽象方法，那么它就是一个函数式接口。对于函数式接口，我们可以通过lambda表达式来创建该接口的对象。

public interface Runnable{
  public abstract void run();
}

常见的函数式接口

1.Runnable

@FunctionalInterface
public interface Runnable {
    public abstract void run();
}

2.Function

@FunctionalInterface
public interface Function<T, R> {
    R apply(T t);
}

3.Consumer

@FunctionalInterface
public interface Consumer<T> {
    void accept(T t);
}

4.Supplier

@FunctionalInterface
public interface Supplier<T> {

    /**
     * Gets a result.
     *
     * @return a result
     */
    T get();
}

5.Biconsumer(Bi代表两个的意思，我们要传入两个参数，在上面的案例中是v和e)

@FunctionalInterface
public interface BiConsumer<T, U> {
    void accept(T t, U u);

}

函数式接口名称	方法名称	参数	返回值
Runnable	run	无参数	无返回值
Function	apply	1个参数	有返回值
Consume	accept	1个参数	无返回值
Supplier	get	没有参数	有返回值
Biconsumer	accept	2个参数	无返回值

链式调用|链式编程|链式写法

public class Chain {
    public static void main(String[] args) {
        //-------------------老式写法------------
//        Student student = new Student();
//        student.setId(1);
//        student.setMajor("cs");
//        student.setName("小卡");
        new Student().setId(1).setName("大卡").setMajor("cs");
    }
    
}

@NoArgsConstructor
@AllArgsConstructor
@Data
@Accessors(chain = true)//开启链式编程
class Student{
    private int id;
    private String name;
    private String major;
}

join和get对比

功能几乎一样，区别在于编码时是否需要抛出异常
- get()方法需要抛出异常
- join()方法不需要抛出异常

public class Chain {
    public static void main(String[] args) throws ExecutionException, InterruptedException {//抛出异常
        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
            return "hello 12345";
        });
        System.out.println(completableFuture.get());
    }

}

public class Chain {
    public static void main(String[] args)  {//抛出异常
        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
            return "hello 12345";
        });
        System.out.println(completableFuture.join());
    }
}

实战精讲-比价网站case

需求

1需求说明
1.1同一款产品，同时搜索出同款产品在各大电商平台的售价；
1.2同一款产品，同时搜索出本产品在同一个电商平台下，各个入驻卖家售价是多少

2输出返回：
出来结果希望是同款产品的在不同地方的价格清单列表，返回一个List
《mysql》in jd price is 88.05
《mysql》in dang dang price is 86.11
《mysql》in tao bao price is 90.43

3解决方案，比对同一个商品在各个平台上的价格，要求获得一个清单列表
1 stepbystep ，按部就班，查完京东查淘宝，查完淘宝查天猫......
2 all in ，万箭齐发，一口气多线程异步任务同时查询。。。

代码实现

/**
 * 案例说明：电商比价需求，模拟如下情况：
 * <p>
 * 1需求：
 * 1.1 同一款产品，同时搜索出同款产品在各大电商平台的售价;
 * 1.2 同一款产品，同时搜索出本产品在同一个电商平台下，各个入驻卖家售价是多少
 * <p>
 * 2输出：出来结果希望是同款产品的在不同地方的价格清单列表，返回一个List<String>
 * 《mysql》 in jd price is 88.05
 * 《mysql》 in dangdang price is 86.11
 * 《mysql》 in taobao price is 90.43
 * <p>
 * 3 技术要求
 * 3.1 函数式编程
 * 3.2 链式编程
 * 3.3 Stream流式计算
 */
public class CompletableFutureMallDemo {
    static List<NetMall> list = Arrays.asList(
            new NetMall("jd"),
            new NetMall("dangdang"),
            new NetMall("taobao"),
            new NetMall("pdd"),
            new NetMall("tmall")
    );

    /**
     * step by step 一家家搜查
     * List<NetMall> ----->map------> List<String>
     *
     * @param list
     * @param productName
     * @return
     */
    public static List<String> getPrice(List<NetMall> list, String productName) {
        //《mysql》 in taobao price is 90.43
        return list
                .stream()
                .map(netMall ->
                        String.format(productName + " in %s price is %.2f",
                                netMall.getNetMallName(),
                                netMall.calcPrice(productName)))
                .collect(Collectors.toList());
    }

    /**
     * List<NetMall> ----->List<CompletableFuture<String>>------> List<String>
     *
     * @param list
     * @param productName
     * @return
     */
    public static List<String> getPriceByCompletableFuture(List<NetMall> list, String productName) {
        return list.stream().map(netMall ->
                CompletableFuture.supplyAsync(() -> String.format(productName + " in %s price is %.2f",
                        netMall.getNetMallName(),
                        netMall.calcPrice(productName))))
                .collect(Collectors.toList())
                .stream()
                .map(s -> s.join())
                .collect(Collectors.toList());
    }


    public static void main(String[] args) {
        long startTime = System.currentTimeMillis();
        //一家家搜查
        List<String> list1 = getPrice(list, "mysql");
        for (String element : list1) {
            System.out.println(element);
        }
        long endTime = System.currentTimeMillis();
        System.out.println("----costTime: " + (endTime - startTime) + " 毫秒");

        System.out.println("--------------------");

        long startTime2 = System.currentTimeMillis();
        //每个都创建一个线程去查找，同步进行
        List<String> list2 = getPriceByCompletableFuture(list, "mysql");
        for (String element : list2) {
            System.out.println(element);
        }
        long endTime2 = System.currentTimeMillis();
        System.out.println("----costTime: " + (endTime2 - startTime2) + " 毫秒");
    }
}

class NetMall {
    @Getter
    private String netMallName;

    public NetMall(String netMallName) {
        this.netMallName = netMallName;
    }

    /**
     * 模拟价格搜索时间
     * @param productName
     * @return
     */
    public double calcPrice(String productName) {
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //返回随机数
        return ThreadLocalRandom.current().nextDouble() * 2 + productName.charAt(0);
    }
}

结果

//逐个查找
mysql in jd price is 110.06
mysql in dangdang price is 110.07
mysql in taobao price is 109.16
mysql in pdd price is 109.22
mysql in tmall price is 109.26
----costTime: 5101 毫秒
--------------------
//开启多个线程查找并获取返回值
mysql in jd price is 109.04
mysql in dangdang price is 110.39
mysql in taobao price is 109.39
mysql in pdd price is 109.01
mysql in tmall price is 110.77
----costTime: 1028 毫秒

CompletableFuture常用API

1.获得结果和触发计算

获取结果
- public T get() 不见不散，容易阻塞
- public T get(long timeout,TimeUnit unit) 过时不候，超过时间会爆异常
- public T join() 类似于get()，区别在于是否需要抛出异常
- public T getNow(T valueIfAbsent)
  - 没有计算完成的情况下，给一个替代结果
  - 立即获取结果不阻塞
    - 计算完，返回计算完成后的结果
    - 没算完，返回设定的valueAbsent(直接返回了备胎值xxx)

public class CompletableFutureAPIDemo {
    public static void main(String[] args) {
        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "abc";
        });

        /**
         * getNow：子线程执行玩需要三秒,主线程执行至此不等待子线程响应,直接获取备胎值
         * 如果在等待时间获取到执行结果则返回
         *     public T getNow(T valueIfAbsent) {
         *         Object r;
         *         return ((r = result) == null) ? valueIfAbsent : reportJoin(r);
         *     }
         */
        TimeUnit.SECONDS.sleep(1);
        String now = completableFuture.getNow("123");
        System.out.println(now);
    }
}
//运行结果123

主动触发计算
- public boolean complete(T value) 是否立即打断get()方法返回括号值
  - (执行要2s，等待只有1s，所以还没执行完就被打断了。返回true表示打断了获取这个过程，直接返回了备胎值complete；如果没打断，返回false 和原来的abc)

public class CompletableFutureAPIDemo2 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "abc";
        });

        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //boolean completableFuture.complete("completeValue") 看是子线程是否执行晚
        //completableFuture.get() 如果子线程没有执行完拿不到返回值，则返回备胎值 "completeValue" 如果子线程拿到了返回值则返回线程返回值
        System.out.println(completableFuture.complete("completeValue")+"\t"+completableFuture.get());
    }
}
//结果
//true	completeValue

2.对计算结果进行处理

thenApply会创建一个新的线程，计算结果存在在依赖关系，使得线程之间串行化。因为依赖关系，所以一旦有异常，直接叫停。

public class CompletableFutureAPIDemo3 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService threadPool = Executors.newFixedThreadPool(3);
        CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
            //暂停几秒钟线程
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("111");
            return 1;
        }, threadPool).thenApply(f -> {
            System.out.println("222");
            int a = 10/0;
            return f + 2;
        }).thenApply(f -> {
            System.out.println("333");
            return f + 3;
        }).whenComplete((v, e) -> {
            if (e == null) {
                System.out.println(v);
            }
        }).exceptionally(e->{
            System.out.println(e.getMessage());
            return null;
        });

    }
}
//正常结果
111
222
333
6
6
//异常结果
111
222
java.lang.ArithmeticException: / by zero
null

handle类似于thenApply，但是有异常的话仍然可以多往下走一步。

public class CompletableFutureAPIDemo4 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService threadPool = Executors.newFixedThreadPool(3);
        CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
            //暂停几秒钟线程
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("111");
            return 1;
        }, threadPool).handle((v, e) -> {
            //模拟异常
            int age = 10/0;
            System.out.println("222");
            return v + 1;
        }).handle((v, e) -> {
            System.out.println("333");
            return v + 3;
        });
        System.out.println(completableFuture.get());
    }
}
//异常结果
111
333
Exception in thread "main" java.util.concurrent.ExecutionException: java.lang.NullPointerException
执行到第三步时 null+3抛出空指针异常

3.对计算结果进行消费

thenRun
- thenRun(Runnable runnable)
- 任务A执行完执行B，并且B不需要A的结果

public class CompletableFutureThenRun {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(3);
        CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
            return 1;
        }, executorService);

        Void run = completableFuture.thenRun(() -> System.out.println("run")).join();
        System.out.println(run);
    }
}

thenAccept
- thenAccept(Consumer action)
- 任务A执行完执行B，B需要A的结果，但是任务B无返回值

接收任务的处理结果，并消费处理，无返回结果|消费型函数式接口，之前的是Function

public static void main(String[] args) throws ExecutionException, InterruptedException {
      ExecutorService executorService = Executors.newFixedThreadPool(3);
      CompletableFuture<Void> voidCompletableFuture = CompletableFuture.supplyAsync(() -> {
            return 1;
        }, executorService).thenAccept(r -> {//消费接口，接收前面线程的返回值
            System.out.println(r + 3);
        }).thenAccept(r->{//前面thenAccept返回值为null,这里接收到的也是null
            System.out.println(r);
        });
        //thenAccept为消费接口
        System.out.println(voidCompletableFuture.get());
}
//运行结果
4
null
null

thenApply
- thenApply(Function fn)
- 任务A执行完执行B，B需要A的结果，同时任务B有返回值

public class CompletableFutureThenApply {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(3);
        CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
            return 1;
        }, executorService).thenApply(r->{
           return r+3;
        }).thenApply(r->{
            return r+3;
        });
        System.out.println(completableFuture.get());
    }
}
//结果
7
//注意：.thenApply需要又连续性才会计算得到7 如果才开进行计算结果为1

4.CompleteFuture和线程池说明（非常重要）

上面的几个方法都有普通版本和后面加Async的版本
以thenRun和thenRunAsync为例，有什么区别？
先看结论

1.没有传入自定义线程池，都用默认线程池ForkJoinPool

2.传入了一个自定义线程池如果你执行第一个任务的时候，传入了一个自定义线程池

调用thenRun方法执行第二个任务的时候，则第二个任务和第一个任务是用同一个线程池
调用thenRunAsync执行第二个任务的时候，则第一个任务使用的是你自己传入的线程池，第二个任务使用的是ForkJoin线程池

3.也有可能处理太快，系统优化切换原则，直接使用main线程处理（把sleep去掉）

public static void main(String[] args) throws ExecutionException, InterruptedException {
    ExecutorService threadPool = Executors.newFixedThreadPool(5);
    CompletableFuture<Void> completableFuture = CompletableFuture.supplyAsync(() -> {
        try {
            TimeUnit.MILLISECONDS.sleep(20);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("1号任务" + "\t" + Thread.currentThread().getName());
        return "abcd";
    }, threadPool).thenRun(() -> {
        try {
            TimeUnit.MILLISECONDS.sleep(20);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("2号任务" + "\t" + Thread.currentThread().getName());
    }).thenRun(() -> {
        try {
            TimeUnit.MILLISECONDS.sleep(20);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("3号任务" + "\t" + Thread.currentThread().getName());
    }).thenRun(() -> {
        try {
            TimeUnit.MILLISECONDS.sleep(20);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("4号任务" + "\t" + Thread.currentThread().getName());
    });
}
//执行结果
1号任务	pool-1-thread-1
2号任务	pool-1-thread-1
3号任务	pool-1-thread-1
4号任务	pool-1-thread-1

public static void main(String[] args) throws ExecutionException, InterruptedException {
    ExecutorService threadPool = Executors.newFixedThreadPool(5);
    CompletableFuture<Void> completableFuture = CompletableFuture.supplyAsync(()->{
        try {TimeUnit.MILLISECONDS.sleep(20);} catch (InterruptedException e) {e.printStackTrace();}
        System.out.println("1号任务"+"\t"+Thread.currentThread().getName());
        return "abcd";
        //thenRunAsync ---这里另起炉灶重新调用了默认的ForkJoinPool
    },threadPool).thenRunAsync(()->{
        try {TimeUnit.MILLISECONDS.sleep(20);} catch (InterruptedException e) {e.printStackTrace();}
        System.out.println("2号任务"+"\t"+Thread.currentThread().getName());
    }).thenRun(()->{
        try {TimeUnit.MILLISECONDS.sleep(20);} catch (InterruptedException e) {e.printStackTrace();}
        System.out.println("3号任务"+"\t"+Thread.currentThread().getName());
    }).thenRun(()->{
        try {TimeUnit.MILLISECONDS.sleep(20);} catch (InterruptedException e) {e.printStackTrace();}
        System.out.println("4号任务"+"\t"+Thread.currentThread().getName());
    });
}
//执行结果
1号任务	pool-1-thread-1
2号任务	ForkJoinPool.commonPool-worker-9---这里另起炉灶重新调用了默认的ForkJoinPool
3号任务	ForkJoinPool.commonPool-worker-9
4号任务	ForkJoinPool.commonPool-worker-9

public class CompletableFutureAPIDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService threadPool = Executors.newFixedThreadPool(5);
        CompletableFuture<Void> completableFuture = CompletableFuture.supplyAsync(()->{
//            try {TimeUnit.MILLISECONDS.sleep(20);} catch (InterruptedException e) {e.printStackTrace();}
            System.out.println("1号任务"+"\t"+Thread.currentThread().getName());
            return "abcd";
        },threadPool).thenRun(()->{
           // try {TimeUnit.MILLISECONDS.sleep(20);} catch (InterruptedException e) {e.printStackTrace();}
            System.out.println("2号任务"+"\t"+Thread.currentThread().getName());
        }).thenRun(()->{
          //  try {TimeUnit.MILLISECONDS.sleep(20);} catch (InterruptedException e) {e.printStackTrace();}
            System.out.println("3号任务"+"\t"+Thread.currentThread().getName());
        }).thenRun(()->{
            //try {TimeUnit.MILLISECONDS.sleep(20);} catch (InterruptedException e) {e.printStackTrace();}
            System.out.println("4号任务"+"\t"+Thread.currentThread().getName());
        });
    }
}
//执行结果，系统优化切换原则，直接使用main线程处理
//1号任务  1号任务  pool-1-thread-1
//2号任务  main
//3号任务  main
//4号任务  main

源码

//CompletableFuture.java 2009行
public CompletableFuture<Void> thenRun(Runnable action) {//传入值是一样的
        return uniRunStage(null, action);
    }

    public CompletableFuture<Void> thenRunAsync(Runnable action) {
        return uniRunStage(asyncPool, action);//但是这里有个异步的线程池asyncPool
    }

//进入asyncPool
private static final boolean useCommonPool =
    (ForkJoinPool.getCommonPoolParallelism() > 1);//一般大于1都是成立的

/**
 * Default executor -- ForkJoinPool.commonPool() unless it cannot
 * support parallelism.
 */
private static final Executor asyncPool = useCommonPool ?
    ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();//所以这里会调用forkJoin线程池

5.对计算速度进行选用

applyToEither方法，快的那个掌权

public static void main(String[] args) throws ExecutionException, InterruptedException
{
    CompletableFuture<String> play1 = CompletableFuture.supplyAsync(() -> {
        System.out.println(Thread.currentThread().getName() + "\t" + "---come in ");
        //暂停几秒钟线程
        try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); }
        return "play1 ";
    });

    CompletableFuture<String> play2 = CompletableFuture.supplyAsync(() -> {
        System.out.println(Thread.currentThread().getName() + "\t" + "---come in ");
        try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); }
        return "play2";
    });
    //applyToEither 对计算速度进行选用,执行快的那一个
    CompletableFuture<String> thenCombineResult = play1.applyToEither(play2, f -> {
        return f + " is winner";
    });

    System.out.println(Thread.currentThread().getName() + "\t" + thenCombineResult.get());
}
//执行结果
ForkJoinPool.commonPool-worker-9	---come in 
ForkJoinPool.commonPool-worker-2	---come in 
main	play2 is winner

6.对计算结果进行合并

thenCombine 合并

两个CompletionStage任务都完成后，最终能把两个任务的结果一起交给thenCOmbine来处理
先完成的先等着，等待其它分支任务

public static void main(String[] args) throws ExecutionException, InterruptedException {
    CompletableFuture<Integer> completableFuture1 = CompletableFuture.supplyAsync(() -> {
        System.out.println(Thread.currentThread().getName() + "\t" + "---come in ");
        return 10;
    });

    CompletableFuture<Integer> completableFuture2 = CompletableFuture.supplyAsync(() -> {
        System.out.println(Thread.currentThread().getName() + "\t" + "---come in ");
        return 20;
    });

    //将两个线程的执行结果进行合并处理
    CompletableFuture<Integer> thenCombineResult = completableFuture1.thenCombine(completableFuture2, (x, y) -> {
        System.out.println(Thread.currentThread().getName() + "\t" + "---come in ");
        //x=10,y=20
        return x + y;
    });
    //30
    System.out.println(thenCombineResult.get());
}

合并版本

public static void main(String[] args) throws ExecutionException, InterruptedException {
    CompletableFuture<Integer> thenCombineResult = CompletableFuture.supplyAsync(() -> {
        System.out.println(Thread.currentThread().getName() + "\t" + "---come in 1");
        return 10;
    }).thenCombine(CompletableFuture.supplyAsync(() -> {
        System.out.println(Thread.currentThread().getName() + "\t" + "---come in 2");
        return 20;
    }), (x, y) -> {
        System.out.println(Thread.currentThread().getName() + "\t" + "---come in 3");
        return x + y;
    }).thenCombine(CompletableFuture.supplyAsync(() -> {
        System.out.println(Thread.currentThread().getName() + "\t" + "---come in 4");
        return 30;
    }), (a, b) -> {
        System.out.println(Thread.currentThread().getName() + "\t" + "---come in 5");
        return a + b;
    });
    System.out.println("-----主线程结束，END");
    System.out.println(thenCombineResult.get());


    // 主线程不要立刻结束，否则CompletableFuture默认使用的线程池会立刻关闭:
    try {
        TimeUnit.SECONDS.sleep(10);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}
//执行结果
ForkJoinPool.commonPool-worker-9	---come in 1
ForkJoinPool.commonPool-worker-9	---come in 2
main	---come in 3
ForkJoinPool.commonPool-worker-9	---come in 4
main	---come in 5
-----主线程结束，END
60

标题：CompletableFuture
作者：llp
地址：https://llinp.cn/articles/2022/11/20/1668947126209.html