Java异步编程之CompletableFuture

ゞ浴缸里的玫瑰 2023-01-13 08:59 161阅读 0赞

CompletableFuture是在 JDK1.8 才加上的类，实现了Future, CompletionStage两个接口，提供了丰富的API，配合流式(stream)编程，让异步任务处理变得简单

*  1.CompletableFuture与Stream结合，处理异步任务
 *  2.CompletableFuture类默认使用ForkJoinPool.commonPool() ，该连接池默认使用CPU核心数为线程数，某些API可接收用户自定义线程池运行任务
 *  3.CompletableFuture提供了一些API可以组合处理不同业务类型的多个异步任务，如常用的：thenAccept、thenApply、whenComplete、thenCombine、allOf等方法

import cn.hutool.core.date.DateUtil;
    
    import java.util.ArrayList;
    import java.util.Arrays;
    import java.util.List;
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.CompletableFuture;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    import java.util.stream.Collectors;
    
    /** * CompletableFuture组合式异步编程 * @description: * @author: zhuyu * @date: 2021/4/10 10:00 */
    public class CompletableFutureUtil { 
    
        private static final int corePoolSize = Runtime.getRuntime().availableProcessors();//CPU核心数
        private static ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(
                corePoolSize,
                corePoolSize * 2,
                1,
                TimeUnit.MINUTES,
                new ArrayBlockingQueue<>(1024),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );
    
        public static void main(String[] args) throws Exception{ 
    
            polling();//重点要理解的方法
            //runAsync();
            //supplyAsync();
    
            //futureComplete();
            //thenAccept();
            //thenApply();
            //whenComplete();
            //thenCombine();
            //allOf();
        }
    
        /** * 模拟网络请求，向20个Ip实例发起请求 * 1.同步方式发起20次网络调用 * 2.CompletableFuture方式发起20次网络调用 */
        public static void polling(){ 
            List<String> ipList = new ArrayList<>();
    
            for(int i=1; i <= 10; i++){ 
                ipList.add("192.168.1." + i);
            }
            //同步调用
            long d1 = System.currentTimeMillis();
            ipList.forEach(ip -> remoteCall(ip));
            long d2 = System.currentTimeMillis();
            System.out.println("sync time : " + (d2 - d1) + " ms");
    
            //异步调用，使用CompletableFuture.supplyAsync异步处理
            List<CompletableFuture<String>> futures = ipList.stream()
                    //将远程调用当成异步任务处理，利用ForkJoinPool.commonPool线程池
                    .map(ip -> CompletableFuture.supplyAsync(() -> remoteCall(ip)))
                    //对异步任务的结果进行加工处理，异步任务完成后，该线程再次对结果进行加工处理
                    .map(future -> future.thenApply(CompletableFutureUtil::convertResult))
                    .collect(Collectors.toList());
            //再使用future.join方法等待每个任务的结果
            List<String> results = futures.stream().map(future -> future.join()).collect(Collectors.toList());
            long d3 = System.currentTimeMillis();
            System.out.println("async time : " + (d3 - d2) + " ms");
    
            for(String str : results){ 
                System.out.println(str);
            }
        }
    
        /** * 对结果再次加工处理 * @param result * @return */
        private static String convertResult(String result){ 
            System.out.println(Thread.currentThread().getName() + " convertResult : " + result);
            return "hello " + result;
        }
    
        /** * 模拟网络调用 * @param param * @return */
        public static String remoteCall(String param){ 
            try { 
                //模拟网络调用，处理业务，睡眠线程1s
                Thread.sleep(1000);
                System.out.println(Thread.currentThread().getName() + " " +DateUtil.now() + " reuqest param: " + param);
            }catch (Exception e){ 
                e.printStackTrace();
            }
            return param;
        }
    
        /** * 运行一个异步任务，无返回结果 * @throws Exception */
        public static void runAsync() throws Exception { 
            CompletableFuture future = CompletableFuture.runAsync(() -> remoteCall("runAsync"));
            future.join(); //同步等待任务处理完成
        }
    
        /** * 运行一个异步任务，获取任务处理结果 * @throws Exception */
        public static void supplyAsync() throws Exception { 
            CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> remoteCall("supplyAsync"));
            System.out.println(DateUtil.now() +  " , response result: " + future.get());
    
            CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> remoteCall("supplyAsync") , poolExecutor);
            System.out.println(DateUtil.now() +  " , response result: " + future2.get());
        }
    
        /** * 创建一个线程，把结果设置到Future中，此方式使用较少 * @throws Exception */
        public static void futureComplete() throws Exception { 
            CompletableFuture<String> future = new CompletableFuture<>();
            new Thread(() -> { 
                try { 
                    //发起网络调用
                    String result = remoteCall("supplyAsync");
                    //把结果设置到future的返回值中
                    future.complete(result);
                }catch (Exception e){ 
                    //设置异常
                    future.completeExceptionally(e);
                }
            }).start();
            System.out.println(DateUtil.now() +  " , response result: " + future.get());
        }
    
        /** * thenAccept(Consumer<T>) 消费上一个任务的结果 * Consumer参数视为上一个任务在完成时要调用的回调函数 * @throws Exception */
        public static void thenAccept() throws Exception{ 
            //第一个任务
            CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> remoteCall("123"));
            //使用thenAccept消费上一个任务的结果
            CompletableFuture<Void> future2 = future.thenAccept(result -> remoteCall("321 " + result));
            future2.join(); //同步阻塞，等待任务处理完成
        }
    
        /** * 如果两个任务有依赖关系，如B依赖A的结果，可以使用thenApply * thenApply(Function<T, U>) 该方法接收一个Function参数，T类型是上一个结点传入，需要被消费的值，U类型是生成的，会被输出的值 * 相当于把上一个异步任务的返回结果转换为 U * @throws Exception */
        public static void thenApply() throws Exception{ 
            CompletableFuture cf = CompletableFuture.supplyAsync(() -> "I'm Awesome")
                    /*这里的msg就是上一个异步任务的返回结果：I'm Awesome*/
                    .thenApply(msg -> String.format("%s and am Super COOL !!!", msg))
                    .thenAccept(result -> System.out.printf("%s\n", result));
    
            CompletableFuture cf1 = CompletableFuture.supplyAsync(()->{ 
                return 5;
            }).thenApply((r)->{ 
                r = r + 1;
                return r;
            });
            System.out.println(cf1.get());
        }
    
        /** * 异步任务执行完时使用 whenComplete 执行回调方法 和捕获异常 exceptionally * 哪怕supplyAsync抛出了异常，whenComplete也会执行 */
        public static void whenComplete(){ 
            CompletableFuture.supplyAsync(()->{ 
                int a = 10/0;
                return 1;
            }).whenComplete((r, e)->{ 
                System.out.println("whenComplete：" + r);
            }).exceptionally(e->{ 
                System.out.println("exceptionally：" + e);
                return 2;
            });
        }
    
        /** * 合并操作 * 合并两个任务的结果，在对其进行统一处理，简言之，这里的回调任务需要等待两个任务都完成后再会触发 */
        public static void thenCombine() throws Exception{ 
            CompletableFuture<String> future1 = CompletableFuture.supplyAsync(()->{ 
                try { 
                    Thread.sleep(3000);
                    System.out.println("future1 invoke ...");
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
                return "future1";
            });
            CompletableFuture<String> future2 = CompletableFuture.supplyAsync(()->{ 
                System.out.println("future2 invoke ...");
                return "future2";
            });
    
            CompletableFuture<String> result = future1.thenCombine(future2, (r1, r2)->{ 
                return "result：" + r1 + r2;
            });
    
            //这里的get是阻塞的，需要等上面两个任务都完成
            System.out.println(result.get());
        }
    
    
        /** * 对于几十个任务都完成后再来执行相应的操作 * 那怎么集中监听所有任务执行结束与否呢？ 使用allOf(CompletableFuture<?>... cfs) */
        public static void allOf(){ 
            long start = System.currentTimeMillis();
            // 结果集
            List<String> list = new ArrayList<>();
    
            List<Integer> taskList = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
            // 全流式处理转换成CompletableFuture[]
            CompletableFuture[] cfs = taskList.stream()
                    .map(integer -> CompletableFuture.supplyAsync(() -> calc(integer))
                            .thenApply(h->Integer.toString(h))
                            .whenComplete((s, e) -> { 
                                System.out.println("任务"+s+"完成!result="+s+"，异常 e="+e+","+ DateUtil.now());
                                list.add(s);
                            })
                    ).toArray(CompletableFuture[]::new);
    
            //监听所有异步任务完成
            CompletableFuture.allOf(cfs).join();
    
            System.out.println("list="+list+",耗时="+(System.currentTimeMillis()-start));
        }
    
        public static int calc(Integer i) { 
            try { 
                if (i == 1) { 
                    Thread.sleep(3000);//任务1耗时3秒
                } else if (i == 5) { 
                    Thread.sleep(5000);//任务5耗时5秒
                } else { 
                    Thread.sleep(1000);//其它任务耗时1秒
                }
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
            return i;
        }
    }