前言future的问题写多线程程序的时候，可以使用future从一个异步线程中拿到结果，但是如果使用过程中会发现一些问题：
如果想要对future的结果做进一步的操作，需要阻塞当前线程
多个future不能被链式的执行，每个future的结果都是独立的，期望对一个future的结果做另外一件异步的事情；
没有异常处理策略，如果future执行失败了，需要手动捕捉
completablefuture应运而生为了解决future问题，jdk在1.8的时候给我们提供了一个好用的工具类completablefuture；
它实现了future和completionstage接口，针对future的不足之处给出了相应的处理方式。
在异步线程执行结束后可以自动回调我们新的处理逻辑，无需阻塞
可以对多个异步任务进行编排，组合或者排序
异常处理
completablefuture的核心思想是将每个异步任务都可以看做一个步骤(completionstage)，然后其他的异步任务可以根据这个步骤做一些想做的事情。
completionstage定义了许多步骤处理的方法，功能非常强大，这里就只列一下日常中常用到的一些方法供大家参考。
使用方式基本使用-提交异步任务简单的使用方式
异步执行，无需结果：
// 可以执行executors异步执行，如果不指定，默认使用forkjoinpoolcompletablefuture.runasync(() -> system.out.println("hello completablefuture!"));
异步执行，同时返回结果：
// 同样可以指定线程池completablefuture<string> stringcompletablefuture = completablefuture.supplyasync(() -> "hello completablefuture!");system.out.println(stringcompletablefuture.get());
处理上个异步任务结果thenrun: 不需要上一步的结果，直接直接新的操作
thenaccept：获取上一步异步处理的内容，进行新的操作
thenapply: 获取上一步的内容，然后产生新的内容
所有加上async后缀的，代表新的处理操作仍然是异步的。async的操作都可以指定executors进行处理
// demo completablefuture .supplyasync(() -> "hello completablefuture!") // 针对上一步的结果做处理，产生新的结果 .thenapplyasync(s -> s.touppercase()) // 针对上一步的结果做处理，不返回结果 .thenacceptasync(s -> system.out.println(s)) // 不需要上一步返回的结果，直接进行操作 .thenrunasync(() -> system.out.println("end")); ;
对两个结果进行选用-accepteither当我们有两个回调在处理的时候，任何完成都可以使用，两者结果没有关系，那么使用accepteither。
两个异步线程谁先执行完成，用谁的结果，其余类型的方法也是如此。
// 返回abccompletablefuture .supplyasync(() -> { sleeputils.sleep(100); return "hello completablefuture!"; }) .accepteither(completablefuture.supplyasync(() -> "abc"), new consumer<string>() { @override public void accept(string s) { system.out.println(s); } });// 返回hello completablefuture! completablefuture .supplyasync(() -> "hello completablefuture!") .accepteither(completablefuture.supplyasync(() -> { sleeputils.sleep(100); return "abc"; }), new consumer<string>() { @override public void accept(string s) { system.out.println(s); } });
对两个结果进行合并-thencombine, thenacceptboththencombine
当我们有两个completionstage时，需要对两个的结果进行整合处理，然后计算得出一个新的结果。
thencompose是对上一个completionstage的结果进行处理，返回结果，并且返回类型必须是completionstage。
thencombine是得到第一个completionstage的结果，然后拿到当前的completionstage，两者的结果进行处理。
completablefuture<integer> heightasync = completablefuture.supplyasync(() -> 172); completablefuture<double> weightasync = completablefuture.supplyasync(() -> 65) .thencombine(heightasync, new bifunction<integer, integer, double>() { @override public double apply(integer wight, integer height) { return wight * 10000.0 / (height * height); } }) ;
thenacceptboth
需要两个异步completablefuture的结果，两者都完成的时候，才进入thenacceptboth回调。
// thenacceptboth案例： completablefuture .supplyasync(() -> "hello completablefuture!") .thenacceptboth(completablefuture.supplyasync(() -> "abc"), new biconsumer<string, string>() { // 参数一为我们刚开始运行时的completablestage，新传入的作为第二个参数 @override public void accept(string s, string s2) { system.out.println("param1=" + s + ", param2=" + s2); } });// 结果：param1=hello completablefuture!, param2=abc
异常处理当我们使用completefuture进行链式调用的时候，多个异步回调中，如果有一个执行出现问题，那么接下来的回调都会停止，所以需要一种异常处理策略。
exceptionally
exceptionally是当出现错误时，给我们机会进行恢复，自定义返回内容。
completablefuture.supplyasync(() -> { throw new runtimeexception("发生错误"); }).exceptionally(throwable -> { log.error("调用错误 {}", throwable.getmessage(), throwable); return "异常处理内容"; });
handle
exceptionally是只有发生异常时才会执行，而handle则是不管是否发生错误都会执行。
completablefuture.supplyasync(() -> { return "abc";}).handle((r,err) -> { log.error("调用错误 {}", err.getmessage(), err); // 对结果做额外的处理 return r;});
案例大量用户发送短信|消息需求为对某个表中特定条件的用户进行短信通知，但是短信用户有成百上千万，如果使用单线程读取效率会很慢。这个时候可以考虑使用多线程的方式进行读取；
1、将读取任务拆分为多个不同的子任务，指定读取的偏移量和个数
// 假设有500万条记录 long recordcount = 500 * 10000; int subtaskrecordcount = 10000; // 对记录进行分片 list<map> subtasklist = new linkedlist<>(); for (int i = 0; i < recordcount / 500; i++) { // 如果子任务结构复杂，建议使用对象 hashmap<string, integer> subtask = new hashmap<>(); subtask.put("index", i); subtask.put("offset", i * subtaskrecordcount); subtask.put("count", subtaskrecordcount); subtasklist.add(subtask); }
2、使用多线程进行批量读取
// 进行subtask批量处理，拆分为不同的任务 subtasklist.stream() .map(subtask -> completablefuture.runasync(()->{ // 读取数据，然后处理 // datatunel.read(subtask); },excuturs)) // 使用应用的通用任务线程池 .map(c -> ((completablefuture<?>) c).join());
3、进行业务逻辑处理，或者直接在读取完进行业务逻辑处理也是可以；
并发获取商品不同信息在系统拆分比较细的时候，价格，优惠券，库存，商品详情等信息分散在不同的系统中，有时候需要同时获取商品的所有信息， 有时候可能只需要获取商品的部分信息。
当然问题点在于要调用多个不同的系统，需要将rt降低下来，那么需要进行并发调用；
list<task> tasklist = new arraylist<>(); list<object> result = tasklist.stream() .map(task -> completablefuture.supplyasync(()->{// handlermap.get(task).query(); return ""; }, executorservice)) .map(c -> c.join()) .collect(collectors.tolist());
问题thenrun和thenrunasync有什么区别如果不使用传入的线程池，大家用默认的线程池forkjoinpool
thenrun用的默认和上一个任务使用相同的线程池
thenrunasync在执行新的任务的时候可以接受传入一个新的线程池，使用新的线程池执行任务；
handle和exceptional有什么区别exceptionally是只有发生异常时才会执行，而handle则是不管是否发生错误都会执行。
以上就是java多线程工具completablefuture怎么使用的详细内容。