JUC并发编程（1）—CompletableFuture详解

⑤. 对计算结果进行合并(thenCombine、thenAcceptBoth、runAfterBoth)

1. public <U,V> CompletableFuture thenCombine(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn)
   两个CompletionStage任务都完成后,最终把两个任务的结果一起交给thenCombine来处理先完成的先等着,等待其他分支任务
   

⑥. 多任务组合(allOf、anyOf)

1. allOf:等待所有任务完成
   (public static CompletableFuture allOf(CompletableFuture<?>... cfs))
2. anyOf:只要有一个任务完成
   (public static CompletableFuture

实战演练

案例说明:电商比价需求
同一款产品,同时搜索出同款产品在各大电商的售价;
同一款产品,同时搜索出本产品在某一个电商平台下,各个入驻门店的售价是多少
出来结果希望是同款产品的在不同地方的价格清单列表,返回一个List
in jd price is 88.05
in pdd price is 86.11
in taobao price is 90.43
(String类型)

代码示例

public class test1 {

    public ExecutorService executorService = new ThreadPoolExecutor(40,100,100,TimeUnit.MINUTES,new ArrayBlockingQueue<>(10000));
    public static List<mall> list = Arrays.asList(
            new mall("jd"),
            new mall("tmall"),
            new mall("taobao"),
            new mall("pdd"),
            new mall("elm")
    );

    public static void main(String[] args) {
        showRes3(list,"mysql");
    }
    public static List<String> showRes(List<mall> list,String productName){
        long start = System.currentTimeMillis();
        List<String> collect = list.stream().map(mall -> String.format(productName + " in %s price is %.2f", mall.getName(), mall.getPrice(productName))).collect(Collectors.toList());
        collect.forEach(System.out::println);
        long end = System.currentTimeMillis();
        System.out.println("耗时："+(end-start)+"秒");
        return collect;
    }
    public static List<String> showRes2(List<mall> list,String productName){
        long start = System.currentTimeMillis();
        List<String> collect = list.parallelStream().map(mall -> CompletableFuture.supplyAsync(() ->
                String.format(productName + " in %s price is %.2f", mall.getName(), mall.getPrice(productName))
        ).join()).collect(Collectors.toList());
        long end = System.currentTimeMillis();
        collect.forEach(System.out::println);
        System.out.println("耗时："+(end-start)+"秒");
        return collect;
    }

    public static List<String> showRes3(List<mall> list,String productName){
        long start = System.currentTimeMillis();
        List<String> collect = list.stream().map(mall -> CompletableFuture.supplyAsync(() -> String.format(productName + " in %s price is %.2f", mall.getName(), mall.getPrice(productName))
        )).collect(Collectors.toList()).stream().map(s -> s.join()).collect(Collectors.toList());
        long end = System.currentTimeMillis();
        collect.forEach(System.out::println);
        System.out.println("耗时："+(end-start)+"毫秒");
        return collect;
    }

}


class mall{
    private String name;

    public String getName() {
        return name;
    }

    public mall(String name) {
        this.name = name;
    }

    public double getPrice(String productName){
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return ThreadLocalRandom.current().nextDouble()*2+productName.charAt(0);
    }
}

通过上面代码示例结果得知
通过异步线程方法耗时需要1000毫秒左右
而同步线程方法耗时需要5000毫秒
而且当商城店铺越来越多的时候，异步线程耗时不会增加，同步线程耗时会不断增加，异步的优势就显现出来了。

tips
在使用stream流的时候，如果使用的是集合的stream（）方法，再使用异步线程时比如使用join()方法会出现串行，和上面方法三一样得使用两次stream流才能实现异步执行，相当于每一步都得使用stream流。
这个时候可以使用parallelStream并行流就可以解决这个问题
但是使用parallelStream会出现一些问题，所以并行流要谨慎使用！！！

并行流的陷阱

1. 线程安全
   由于并行流使用多线程，则一切线程安全问题都应该是需要考虑的问题，如：资源竞争、死锁、事务、可见性等等。
2. 线程消费
   在虚拟机启动时，我们指定了worker线程的数量，整个程序的生命周期都将使用这些工作线程；这必然存在任务生产和消费的问题，如果某个生产者生产了许多重量级的任务（耗时很长），那么其他任务毫无疑问将会没有工作线程可用；更可怕的事情是这些工作线程正在进行IO阻塞。