CompletableFuture使用教学

一、开始一个线程异步执行不需要返回值

通过runAsync方式

//1.不加线程池方式
CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(() -> {
            System.out.println(Thread.currentThread().getName());
            //停顿几秒
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
        System.out.println("hello world");
        System.out.println(completableFuture.get());//null 没有返回值的情况

//2.加线程池方式
		//创建固定线程池（阿里规范建议使用自定义线程池，不能通过Executors来进行创建）
        ExecutorService executors = Executors.newFixedThreadPool(4);//此处偷懒，用此线程池
        CompletableFuture<Void> runAsync = CompletableFuture.runAsync(() -> {
            System.out.println(Thread.currentThread().getName());
            //停顿几秒
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }, executors);
        System.out.println(runAsync.get());

二、通过异步方式执行，有返回值

supplyAsync

		//不加线程池方式
        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName());
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return "hello supplyAsync";
        });
        System.out.println(completableFuture.get());

		//加线程池的方式
ExecutorService executors = Executors.newFixedThreadPool(4);
        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName());
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return "hello supplyAsync+Executors";
        }, executors);
        System.out.println(completableFuture.get());

ps：get（）方法可以获取异步线程执行完后的结果

三.通过whenComplete减少阻塞和轮询(可加线程池,也可不加)

即当异步线程执行结束会接着执行whenComplete（）方法，如果执行期间报错会执行exceptionally()方法

ExecutorService threadPool = Executors.newFixedThreadPool(6);
        CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "--副线程");
            int result = new Random().nextInt(10);
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            System.out.println("1秒后出结果");
            return result;
        },threadPool).whenComplete((v,e)->{//没有异常  v是值 e是异常情况
            if (e == null){
                System.out.println("计算完成,UpdateValue:"+v);
            }
        }).exceptionally((e)->{//e是异常情况

            e.printStackTrace();
            System.out.println("异常情况:"+e.getCause()+"\t"+e.getMessage());
            return null;
        });

四、实现通过货品在不同平台进行价格搜索进行数据汇总

普通方式实现

public class Case {
    static List<NetMall> list = Arrays.asList(new NetMall("jd"),
            new NetMall("dangdang"),
            new NetMall("taobao"));
    public static List<String> getPrice(List<NetMall> list,String productName){
        return list.stream()
                .map(netMall -> String.format(productName+ " in %s price is %s"
                        ,netMall.getNetMallName()
                        ,netMall.calcPrice(productName)))
                .collect(Collectors.toList());
    }

    public static void main(String[] args) {
        long start = System.currentTimeMillis();
        List<String> list1 = getPrice(list, "mysql");
        for (String s : list1) {
            System.out.println(s);
        }
        long end = System.currentTimeMillis();
        System.out.println("---当前操作时间--costTime:"+(end - start)+"ms");
    }

}
class NetMall{
    private String netMallName;

    public String getNetMallName() {
        return netMallName;
    }
    public NetMall(){}

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

    public double calcPrice(String productName){
        try {
            TimeUnit.SECONDS.sleep(1);//此处表示 业务执行所需耗时时间
        }catch (InterruptedException e){
            e.printStackTrace();
        }
        return ThreadLocalRandom.current().nextDouble() * 2 + productName.charAt(0);//模拟价格
    }

}

普通方式实现耗时结果：

completableFuture实现

public class Case {
    static List<NetMall> list = Arrays.asList(new NetMall("jd"),
            new NetMall("dangdang"),
            new NetMall("taobao"));
    public static List<String> getPricesByCompletableFuture(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 start = System.currentTimeMillis();
		List<String> list1 = getPricesByCompletableFuture(Case.list, "mysql");
        for (String s : list1) {
            System.out.println(s);
        }
        long end = System.currentTimeMillis();
        System.out.println("---当前操作时间--costTime:"+(end - start)+"ms");
    }

}
class NetMall{
    private String netMallName;

    public String getNetMallName() {
        return netMallName;
    }
    public NetMall(){}

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

    public double calcPrice(String productName){
        try {
            TimeUnit.SECONDS.sleep(1);
        }catch (InterruptedException e){
            e.printStackTrace();
        }
        return ThreadLocalRandom.current().nextDouble() * 2 + productName.charAt(0);//模拟价格
    }

}

结果耗时：

相比较能够发现，同时开启三个异步线程，时间仅仅为单个平台查询的时间，大大节省效率！

五、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 boolean complete(T value) 是否立即打断get()方法返回括号值（下面代码实现）

public static void main(String[] args) throws InterruptedException, ExecutionException {
        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return "abc";
        });
        TimeUnit.SECONDS.sleep(1);
        System.out.println(completableFuture.complete("end")+"\t"+completableFuture.get());
    }

执行结果：true end
解释：执行需要2秒，等待1秒；
complete（默认值）方法会打断执行，如果执行完，则返回结果，如果没有执行完则输出默认值；

2.对计算结果进行处理

（1）thenApply 计算结果存在在依赖关系，使得线程串行化。因为依赖关系，所以一旦有异常，直接叫停。

public static void main(String[] args) {
        CompletableFuture.supplyAsync(()->{
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("111");
            return 1024;
        }).thenApply(f->{
            System.out.println("222");
            return f + 1;
        }).thenApply(f->{
            System.out.println("333");
            return f + 1;
//            return f/0; 会报出异常
        }).whenCompleteAsync((v,e)->{
            System.out.println("****v="+v);
        }).exceptionally(e->{
            e.printStackTrace();
            return null;
        });
        System.out.println("----主线程结束--end");
        // 主线程不要立刻结束，否则CompletableFuture默认使用的线程池会立刻关闭:
        try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); }
    }

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

CompletableFuture.supplyAsync(() -> {
            //暂停几秒钟线程
            try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); }
            System.out.println("111");
            return 1024;
        }).handle((f,e) -> {
            int age = 10/0;//异常语句
            System.out.println("222");
            return f + 1;
        }).handle((f,e) -> {
            System.out.println("333");
            return f + 1;
        }).whenCompleteAsync((v,e) -> {
            System.out.println("*****v: "+v);
        }).exceptionally(e -> {
            e.printStackTrace();
            return null;
        });
        System.out.println("----主线程结束--end");
        //在222方法中报错 会继续执行输出333

（3）.对计算结果进行消费
接收任务的处理结果，并消费处理，无返回结果|消费型函数式接口，之前的是Function

CompletableFuture.supplyAsync(()->{
            return 1;
        }).thenApply(f -> {
            return f+1;
        }).thenApply(f ->{
            return f+3;
        }).thenApply(f->{
            return f+4;
        }).thenAccept(r->{
            System.out.println("r==\t"+r);
        });

补充：Code之任务之间的顺序执行

thenRun

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

thenAccept

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

thenApply

• thenApply(Function fn)

• 任务A执行完执行B，B需要A的结果，同时任务B有返回值

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

上面的几个方法都有普通版本和后面加Async的版本
以thenRun和thenRunAsync为例，有什么区别？先看结论
1.没有传入自定义线程池，都用默认线程池ForkJoinPool
2.传入了一个自定义线程池如果你执行第一个任务的时候，传入了一个自定义线程池

• 调用thenRun方法执行第二个任务的时候，则第二个任务和第一个任务是用同一个线程池
• 调用thenRunAsync执行第二个任务的时候，则第一个任务使用的是你自己传入的线程池，第二个任务使用的是ForkJoin线程池
  3.也有可能处理太快，系统优化切换原则，直接使用main线程处理（把sleep去掉）

（5）.对计算速度进行选用

//applyToEither 线程先执行完的输出，输出其中一个
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(3); } catch (InterruptedException e) { e.printStackTrace(); }
            return "play2 ";
        });
        CompletableFuture<String> thenCombineResult  = play1.applyToEither(play2, f -> {
            return f + "is winner";
        });
        System.out.println(Thread.currentThread().getName()+"\t"+thenCombineResult.get());

（6）.对计算结果进行合并‘

thenCombine 合并

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

        CompletableFuture<Integer> completeFuture1 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "\t" + "---come in");
            return 10;
        });
        CompletableFuture<Integer> completeFuture2 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "\t" + "---come in");
            return 20;
        });
        CompletableFuture<Integer> completableFuture = completeFuture1.thenCombine(completeFuture2,
                (x, y) -> {
            System.out.println(Thread.currentThread().getName() + "\t" + "---come in");
            return x + y;
        });
        System.out.println(completableFuture.get()); //输出 30

合并版本

CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "\t" + "---come in1");
            return 10;
        }).thenCombine(CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "\t" + "---come in2");
            return 20;
        }), (x, y) -> {
            System.out.println(Thread.currentThread().getName() + "\t" + "---come in3");
            return x + y;
        }).thenCombine(CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "\t" + "---come in4");
            return 30;
        }), (x, y) -> {
            System.out.println(Thread.currentThread().getName() + "\t" + "---come in5");
            return x + y;
        });
        System.out.println(completableFuture.get());
        //输出结果
        //ForkJoinPool.commonPool-worker-9	---come in1
		//ForkJoinPool.commonPool-worker-9	---come in2
		//main	---come in3
		//ForkJoinPool.commonPool-worker-2	---come in4
		//main	---come in5
		//60