一、背景

jdk8中加入了实现类CompletableFuture，用于异步编程。底层做任务使用的是ForkJoin, 顾名思义，是将任务的数据集分为多个子数据集，而每个子集，都可以由独立的子任务来处理，最后将每个子任务的结果汇集起来。它是ExecutorService接口的一个实现，它把子任务分配给线程池（称为ForkJoinPool）中的工作线程。从api文档看，它实现了2个接口CompletionStage和Future。CompletionStage支持lambda表达式，接口的方法的功能都是在某个阶段得到结果后要做的事情。因此，CompletableFuture不仅拥有Future的所有特性，而且还内置了lambda表达式，支持异步回调，结果转换等功能，它有以下Future实现不了的功能：

  1. 合并两个相互独立的异步计算的结果

  2. 等待异步任务的所有任务都完成

  3. 等待异步任务的其中一个任务完成就返回结果

  4. 任务完成后调用回调方法

  5. 任务完成的结果可以用于下一个任务。

  6. 任务完成时发出通知提供原生的异常处理api

二、代码

package com.example.demo;

import java.util.ArrayList;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.concurrent.*;

public class CompletableFutureDemo {

     //CPU核数

    private static final int AVAILABLE_PROCESSORS = Runtime.getRuntime().availableProcessors();

    private static final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(AVAILABLE_PROCESSORS,

            3 * AVAILABLE_PROCESSORS,

            3, TimeUnit.SECONDS,

            new LinkedBlockingDeque<>(20));

    public static void main(String[] args) throws Exception {

        long startTime = System.currentTimeMillis();

        System.out.println("demo start....." + startTime);

        demo3();

        System.out.println("demo end.....costTime = " + (System.currentTimeMillis() - startTime));

    }

    /**

     * 基于allOf，并行处理多个任务，等待所有任务执行完毕后返回

     */

    public static void demo3() throws Exception {

       //用户整体接收各个任务的返回值

        Map<String,String> dataMap = new ConcurrentHashMap<>();

        List<CompletableFuture<String>> futureList = new ArrayList<>();

        futureList.add(doSomethingA("A", dataMap));

        futureList.add(doSomethingB("B", dataMap));

        futureList.add(doSomethingC("C", dataMap));

        CompletableFuture<Void> result = CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0]));

        try {

                result.get(3, TimeUnit.SECONDS);

        } catch (Exception e) {

            e.printStackTrace();

        }

        System.out.println("dataMap = " + dataMap);

       //结果为：{doSomeThingB=B, doSomeThingA=A}

    }

    /**

     * 基于thenCompose，第一个任务执行完后，第二个任务使用第一个任务的返回作为参数

     */

    public static void demo1() throws Exception {

        Map<String,String> dataMap = new HashMap<>();

        CompletableFuture<String> completableFuture = doSomethingA("A", dataMap)

                .thenCompose(id -> doSomethingB(id, dataMap));

        String result = completableFuture.get(3, TimeUnit.SECONDS);

        System.out.println("result = " + result);

        //结果为：A is done is done

    }

    /**

     * 基于thenCombine，当两个任务都完成后，使用两者的结果作为参数再执行一个异步任务

     */

    public static void demo2() throws Exception {

        Map<String,String> dataMap = new HashMap<>();

        CompletableFuture<String> completableFuture = doSomethingA("A", dataMap)

                .thenCombine(doSomethingB("B", dataMap), (a, b) -> a + " - " + b);

        String result = completableFuture.get(3, TimeUnit.SECONDS);

        System.out.println("result = " + result);

//结果为：A is done - B is done

    }

    /**

     * @param dataMap 用户整体接收方法的返回值

     * @return

     */

    public static CompletableFuture<String> doSomethingA(String taskId, Map<String,String> dataMap) {

        System.out.println("doSomethingA start....." + System.currentTimeMillis());

        return CompletableFuture.supplyAsync(() -> {

            try {

                Thread.sleep(100);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            dataMap.put("doSomeThingA", "A");

            System.out.println(taskId + " is done and dataMap"+dataMap);

            return taskId + " is done";

        }, threadPoolExecutor);

    }

    public static CompletableFuture<String> doSomethingB(String taskId, Map<String,String> dataMap) {

        System.out.println("doSomethingB start....." + System.currentTimeMillis());

        return CompletableFuture.supplyAsync(() -> {

            try {

                Thread.sleep(100);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            dataMap.put("doSomeThingB", "B");

            System.out.println(taskId + " is done and dataMap"+dataMap);

            return taskId + " -> B is done";

        }, threadPoolExecutor);

    }

    public static CompletableFuture<String> doSomethingC(String taskId, Map<String,String> dataMap) {

        System.out.println("doSomethingC start....." + System.currentTimeMillis());

        return CompletableFuture.supplyAsync(() -> {

            try {

                Thread.sleep(500);

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

            dataMap.put("doSomeThingC", "C");

            System.out.println(taskId + " is done and dataMap"+dataMap);

            return taskId + " is done";

        }, threadPoolExecutor);

    }

}

三、效率比较

很明显，异步更快

package com.example.demo;

import java.util.ArrayList;

import java.util.Arrays;

import java.util.List;

import java.util.concurrent.CompletableFuture;

/**

 * @author d00018641

 * @date 2021/11/4 15:10

 */

public class TestDemo2 {

    private static final String key = "llllllllllllllllllllllll";

    public static void main(String[] args) {

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

        requestList.add("3");

        requestList.add("4");

        requestList.add("5");

        requestList.add("6");

        // 响应参数list

        String[] returnArray = new String[requestList.size()];

        // 异步查询每一列,定义响应列数的futures

        List<CompletableFuture<String>> futures = new ArrayList<>();

        long startTime = System.currentTimeMillis();

        for (int i = 0; i < requestList.size(); i++) {

            final int a = i;

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

                return calc(requestList.get(a));

            }).whenComplete((m, e) -> returnArray[a] = m);

            futures.add(tf);

        }

        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

        //CompletableFuture end.....costTime = 147

        System.out.println("CompletableFuture end.....costTime = " + (System.currentTimeMillis() - startTime));

        long startTime1 = System.currentTimeMillis();

        for(int i = 0; i < requestList.size(); i++){

            returnArray[i] = calc(requestList.get(i));

        }

        //连续 end.....costTime = 432

        System.out.println("连续 end.....costTime = " + (System.currentTimeMillis() - startTime1));

        System.out.println(Arrays.asList(returnArray));

    }

    private static String calc(String source) {

        int as = Integer.parseInt(source);

        try {

            Thread.sleep(100);

        } catch (InterruptedException e) {

            e.printStackTrace();

        }

        return String.valueOf(Math.pow(as, 3));

    }

}

四、编程实战

Map<String, ExchangeRateVO> cacheMap = new ConcurrentHashMap<>();

        List<CompletableFuture<Void>> batchFutureList = new ArrayList<>();

        for (ExchangeRateVO vo : paramList) {

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

                List<CompletableFuture<Void>> itemFutureList = new ArrayList<>();

                for (LookupItemVO toCurrency : toCurrencyList) {

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

                        // 查询接口返回汇率数据

                        ExchangeRateVO resultVo;

                        try {

                            buildBasicContext();

                            resultVo = iExchangeRateService.findExchangeRate(vo.getCountryCode(), vo.getFromCurrency(),

                                toCurrency.getItemCode(), vo.getStartDate());

                            if (CommonUtils.isNotEmpty(resultVo)) {

                                if (CommonUtils.isEmpty(resultVo.getFromCurrency())) {

                                    resultVo.setFromCurrency(vo.getFromCurrency());

                                    resultVo.setToCurrency(toCurrency.getItemCode());

                                    resultVo.setCountryCode(vo.getCountryCode());

                                }

                                resultVo.setStartDate(vo.getStartDate());

                                resultList.add(resultVo);

                                cacheMap.put(vo.getCountryCode() + vo.getFromCurrency() + toCurrency.getItemCode(),

                                    resultVo);

                            }

                        } catch (Exception er) {

                            LOGGER.error("findExchangeRate has some error:", er);

                        }

                    });

                    itemFutureList.add(cfItem);

                }

                CompletableFuture.allOf(itemFutureList.toArray(new CompletableFuture[itemFutureList.size()])).join();

            });

            batchFutureList.add(batchItem);

        }

        LOGGER.info2("before submit completable");

        CompletableFuture.allOf(batchFutureList.toArray(new CompletableFuture[batchFutureList.size()])).join();

最新文章

  1. iOS内支付
  2. 用GCD线程组与GCD信号量将异步线程转换为同步线程
  3. python异常处理（基础）
  4. 学习笔记——Maven 内置变量
  5. struts2文件上传类型的过滤
  6. JAVA static 作用
  7. 原生js的兼容问题总结
  8. HDU - 1394 Minimum Inversion Number （线段树求逆序数）
  9. html5中viewport与meta详解
  10. linux命令详解之df命令
  11. 网络编程-SOCKET开发之----2. TCP粘包现象产生分析
  12. CSS-图片占位的技巧
  13. fft ocean注解
  14. 驱动链表（LIST_ENTRY）
  15. 实现网络数据提取你需要哪些java知识
  16. 说说Python程序的执行过程
  17. 【RF库Collections测试】Remove Values From List
  18. 7 无线wifi传输视频开发
  19. bzoj 1337 最小圆覆盖
  20. linux 提示符绝对路径

热门文章

  1. 一条sql同时返回多个count结果
  2. Gitlab Ubuntu部署
  3. 067_VFPage中Js与controller交互方式（二） RemoteAction
  4. QML与python互相通信
  5. PyCharm如何实现控制台换行显示
  6. SpringBoot Circular view path错误
  7. ElasticSearch (Es) 分组查询 记录
  8. Jquery 简单实现demo
  9. 来自StateOfJS调查：2022年JavaScript生态圈趋势报告(前端早读课)
  10. 直播平台搭建源码，canvas 画一条波浪线 进度条