1.异步执行无返回值

 CompletableFuture noArgsFuture = CompletableFuture.runAsync(new Runnable() {
            @Override
            public void run() {
                System.out.println("***************");
            }
        });

2.异步执行有返回值

 CompletableFuture<String> returnString = CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                return "abc";
            }
        });
        String result = returnString.get();
        System.out.println(result);

3.多个异步执行，等待所有结果返回才算完成

 CompletableFuture<String> feature1 = CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                try {
                    Thread.sleep(3000L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return "abc";
            }
        });
        CompletableFuture<String> feature2 = CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                return "def";
            }
        });
        CompletableFuture<String> feature3 = CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                return "ghi";
            }
        });
        CompletableFuture totalFeature = CompletableFuture.allOf(feature1, feature2, feature3);
        totalFeature.join();
        String str3=feature3.get();
        String str2=feature2.get();
        String str1=feature1.get();
        List<String> stringList = Stream.of(feature1, feature2, feature3).map(CompletableFuture::join).collect(Collectors.toList());
        System.out.println(stringList);
        System.out.println(str1+str2+str3);

4.多个异步执行，有参数有返回值

        //模拟入参
        List<Date> dates = new ArrayList<>();
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        try {
            dates.add(sdf.parse("2020-02-22 12:12:01"));
            dates.add(sdf.parse("2020-02-23 12:12:01"));
            dates.add(sdf.parse("2020-02-24 12:12:01"));
        } catch (ParseException e) {
            e.printStackTrace();
        }
        List<CompletableFuture<String>> futures = new ArrayList<>();
        //将任务提交，并异步执行
        for (Date date : dates) {
            CompletableFuture<String> newFeature = CompletableFuture.supplyAsync(new Supplier<String>() {
                @Override
                public String get() {
                    //异步处理逻辑，这里将日期转换为字符串
                    return coverLongToString(date);
                }
            });
            futures.add(newFeature);
        }
        //所有任务执行完成返回才会继续，不然会阻塞在这里等待
        //TODO 这里怎么能够设置下超时时间
        CompletableFuture.allOf(futures.iterator().next());
        //拿到所有异步线程返回的结果
        List<String> dateStringList = futures.stream().map(CompletableFuture::join).collect(Collectors.toList());
        System.out.println(dateStringList);

java多线程可以直接通过ExecutorService接口的execute方法提交任务，也可以通过invokeAll方法让多个任务一起提交，前者execute是异步提交，也就是说不会阻塞主线程，后则会阻塞主线程，等待所有任务执行完成主线程才回继续执行。其实不管是execute还是invokeAll，它们最终都会去执行Executor接口的execute（Runnable run）方法，下面是截取的部分源码

很明显invokeAll方法调用了execute，并且传入的是一个继承自Runnable接口的RunnableFuture对象，所以其实它们的原理是一样的， 还有一点注意到了，这里还调用了Future对象的get，这个方法是阻塞性的，这也是为什么invokeAll会阻塞主线程的原因。

背景

最近遇到多数据库执行sql处理后生成报表，项目原有逻辑是单线程同步执行的，随着数据量的增大时间越来越长，我忍不住要优化。考虑一定得用多线程处理，那就首先需要线程池了，毕竟没有线程池的情况下很容易出现无休止创建线程是非常危险的。

实现

代码

伪代码

//创建线程池
ExecutorService executor = Executors.newFixedThreadPool(4);
CompletionService<Object> completionSevice = new ExecutorCompletionService<Object>(executor);

for (DatabaseNode databaseNode: databaseNodeList) {
    //提交请求到具体的库
     completionSevice.submit(new SqlHandler(properties, databaseNode, sql,         
     ModuleConstant.NODE_PASSWORD_KEY, encode, userType));

}


//获取执行结果
//databaseNodeList.size()只为获取节点个数，获取到的每个节点是没有顺序
 for (int i = 0; i < databaseNodeList.size(); i++) {
    Object result = null;
    DrdsNodes nodeSchema = null;
    //获取到每个库节点执行的结果，结果是从最先执行完的开始拿的，每次拿一个
    Map<DrdsNodes, Object> resultMap = (Map<DatabaseNode , Object>) 
          completionSevice.take().get();
  //后续根据具体业务操作
   。。。。

 }


/**
  *具体执行sql的类  （伪代码）
  *
 */
public class SqlHandler  implements Callable<Object> {
	
	private final String ddl;
	private JdbcTemplate jdbcTemplate;
    private DrdsNodes databaseNode;
	
	public SqlRowNumHandler(RdsProperties properties, DatabaseNode databaseNode, String ddl, String nodePwdKey,String encode, Integer userType) {
		
       //sql的构建封装代码

	}
	
   //类实现Callable<Object>，初始化时自动调用此方法
	@Override
    public Object call() throws Exception {
       //key 是数据库的实例，用以获取时区分是哪个库执行的结果
        Map<DrdsNodes, Object> resultMap = Maps.newHashMap();
        try {
            Object value = jdbcTemplate.update(ddl);
            resultMap.put(drdsNode, value);
            return resultMap;
        } catch (Exception e) {
            resultMap.put(drdsNode, e.getMessage());
            return resultMap;
        }
    }

注：completionSevice.take().get();是类似于指针一样获取结果（先执行完返回的先获取到）

CompletionService实现原理

CompletionService实际上可以看做是Executor和BlockingQueue的结合体。CompletionService在接收到要执行的任务时，通过类似BlockingQueue的put和take获得任务执行的结果。CompletionService的一个实现是ExecutorCompletionService，ExecutorCompletionService把具体的计算任务交给Executor完成。

测试后发现效率提高了5倍左右。