标签：return System 任务详解 CompletableFuture println out

CompletableFuture

前言

　　CompletableFuture继承于java.util.concurrent.Future，它本身具备Future的所有特性，并且基于JDK1.8的流式编程以及Lambda表达式等实现一元操作符、异步性以及事件驱动编程模型，可以用来实现多线程的串行关系，并行关系，聚合关系。它的灵活性和更强大的功能是Future无法比拟的。

一、创建方式

1. 用默认线程池

　　CompletableFuture<String> future = new CompletableFuture<>();

　　默认使用 ForkJoinPool.commonPool()，commonPool是一个会被很多任务共享的线程池，比如同一 JVM 上的所有 CompletableFuture、并行 Stream 都将共享 commonPool，commonPool 设计时的目标场景是运行非阻塞的 CPU 密集型任务，为最大化利用 CPU，其线程数默认为 CPU 数量 - 1。

2. 用自定义线程池

java 复制代码

ThreadPoolExecutor pool = new ThreadPoolExecutor(2, 4, 3,
                TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(3),
                new ThreadPoolExecutor.DiscardOldestPolicy());
CompletableFuture.runAsync(() -> System.out.println("Hello World!"), pool);

二、使用示例

1. 构建异步任务

方法	有无返回值	描述
runAsync	无	进行数据处理，接收前一步骤传递的数据，无返回值。
supplyAsync	有	进行数据处理，接收前一步骤传递的数据，处理加工后返回。返回数据类型可以和前一步骤返回的数据类型不同。

（1）runAsync

源码

java 复制代码

   public static CompletableFuture<Void> runAsync(Runnable runnable) {
        return asyncRunStage(asyncPool, runnable);
    }

示例

java 复制代码

public static void runAsync() {
    //使用默认线程池
    CompletableFuture cf = CompletableFuture.runAsync(() -> System.out.println("Hello World!"));
    assertFalse(cf.isDone());    
    //使用自定义线程池
    CompletableFuture.runAsync(() -> System.out.println("Hello World!"),                                                 Executors.newSingleThreadExecutor());
    }

（2）supplyAsync

源码

java 复制代码

public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
    return asyncSupplyStage(asyncPool, supplier);
}

示例

java 复制代码

public static void supplyAsync() throws ExecutionException, InterruptedException {
        CompletableFuture<String> f = CompletableFuture.supplyAsync(() -> {
            try {
                //ForkJoinPool.commonPool-worker-1线程
                System.out.println(Thread.currentThread().getName());
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "hello";
        });
        //阻塞等待3秒 
        String result = f.get();
        //main线程
        System.out.println(Thread.currentThread().getName());
        System.out.println(result);
}

2. 单任务结果消费

方法	有无返回值	描述
thenApply	有	在前一个阶段上应用thenApply函数，将上一阶段完成的结果作为当前阶段的入参
thenAccept	无返回值	消费前一阶段的结果
thenRun	无返回值，并且无入参	当上一阶段完成后，执行本阶段的任务

（1）thenApply

源码

java 复制代码

public <U> CompletableFuture<U> thenApplyAsync(
    Function<? super T,? extends U> fn) {
    return uniApplyStage(asyncPool, fn);
}

示例

csharp 复制代码

public static void thenApply() throws ExecutionException, InterruptedException {
    CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(s -> {
        System.out.println(s);
        return s.toUpperCase();
    }).thenApply(s->{
        System.out.println(s);
        return s + ":body";
    });
    System.out.println(cf.get());
}

then意味着这个阶段的动作发生当前的阶段正常完成之后。本例中，当前节点完成，返回字符串message。

Apply意味着返回的阶段将会对结果前一阶段的结果应用一个函数。

函数的执行会被阻塞

（2）thenAccept

源码

scss 复制代码

public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
    return asyncSupplyStage(asyncPool, supplier);
}

示例

java 复制代码

public static void thenAccept() throws InterruptedException {
    CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
        return "message";
    }).thenAccept((consumer) -> {
        System.out.println(consumer);
    });
}

（3）thenRun

源码

java 复制代码

public CompletableFuture<Void> thenRun(Runnable action) {
    return uniRunStage(null, action);
}

示例

java 复制代码

public static void thenRun() throws InterruptedException {
    CompletableFuture.supplyAsync(() -> {
        //执行异步任务
        System.out.println("执行任务");
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "success";
    }).thenRun(() -> {
        // Computation Finished.
        System.out.println("上一阶段任务执行完成");
    });
    Thread.sleep(2000);
}

3. 合并结果消费

方法	有无返回值	描述
thenCombine	有	合并另外一个任务，两个任务都完成后，执行BiFunction，入参为两个任务结果，返回新结果
thenAcceptBoth	无	合并另外一个任务，两个任务都完成后，执行这个方法等待第一个阶段的完成(大写转换)，它的结果传给一个指定的返回CompletableFuture函数，它的结果就是返回的CompletableFuture的结果，入参为两个任务结果，不返回新结果
runAfterBoth	无返回值无入参	合并另外一个任务，两个任务都完成后，执行Runnable，注意，这里的两个任务是同时执行

（1）thenCombine

　　如果CompletableFuture依赖两个前面阶段的结果，它复合两个阶段的结果再返回一个结果，我们就可以使用thenCombine()函数。整个流水线是同步的。

源码

java 复制代码

public <U,V> CompletableFuture<V> thenCombine(CompletionStage<? extends U> other,
                             BiFunction<? super T,? super U,? extends V> fn) {
    return biApplyStage(null, other, fn);
}

示例

java 复制代码

public static void thenCombine() {
    CompletableFuture<String> cfA = CompletableFuture.supplyAsync(() -> {
        System.out.println("processing a...");
        return "hello";
    });
    CompletableFuture<String> cfB = CompletableFuture.supplyAsync(() -> {
        System.out.println("processing b...");
        return " world";
    });
    CompletableFuture<String> cfC = CompletableFuture.supplyAsync(() -> {
        System.out.println("processing c...");
        return ", I'm CodingTao!";
    });
    cfA.thenCombine(cfB, (resultA, resultB) -> {
        System.out.println(resultA + resultB);  // hello world
        return resultA + resultB;
    }).thenCombine(cfC, (resultAB, resultC) -> {
        System.out.println(resultAB + resultC); // hello world, I'm CodingTao!
        return resultAB + resultC;
    });
}

（2）thenAcceptBoth

源码

java 复制代码

public <U> CompletableFuture<Void> thenAcceptBoth(CompletionStage<? extends U> other,
                          BiConsumer<? super T, ? super U> action) {
    return biAcceptStage(null, other, action);
}

示例

java 复制代码

private static void thenAcceptBoth() {
    CompletableFuture<String> cfA = CompletableFuture.supplyAsync(() -> "resultA");
    CompletableFuture<String> cfB = CompletableFuture.supplyAsync(() -> "resultB");
    cfA.thenAcceptBoth(cfB, (resultA, resultB) -> {
        //resultA,resultB
        System.out.println(resultA+","+resultB);
    });
}

（3）runAfterBoth

源码

java 复制代码

public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other,
                                            Runnable action) {
    return biRunStage(null, other, action);
}

示例

java 复制代码

private static void runAfterBoth() {
    CompletableFuture<String> cfA = CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("process a");
        return "resultA";
    });
    CompletableFuture<String> cfB = CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("process b");
        return "resultB";
    });
    cfB.runAfterBoth(cfA, () -> {
        //resultA,resultB
        System.out.println("任务A和任务B同时完成");
    });
    try {
        Thread.sleep(6000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

4. 任一结果消费

方法	有无返回值	描述
applyToEither	有	其中任一任务完成后，执行Function，结果转换，入参为已完成的任务结果。返回新结果，要求两个任务结果为同一类型
acceptEither	无	其中任一任务完成后，执行Consumer，消费结果，入参为已完成的任务结果。不返回新结果，要求两个任务结果为同一类型
runAfterEither	无返回值无入参	其中任一任务完成后，执行Runnable，消费结果，无入参。不返回新结果，不要求两个任务结果为同一类型

场景

　　假设查询商品a，有两种方式，A和B，但是A和B的执行速度不一样，希望哪个先返回就用那个的返回值。

（1）applyToEither

源码

java 复制代码

public <U> CompletableFuture<U> applyToEither(CompletionStage<? extends T> other,                                                  Function<? super T, U> fn) {
    return orApplyStage(null, other, fn);
}

示例

java 复制代码

private static void applyToEither() throws ExecutionException, InterruptedException {
    CompletableFuture<String> futureA = CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "通过方式A获取商品a";
    });
    CompletableFuture<String> futureB = CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "通过方式B获取商品a";
    });
    CompletableFuture<String> futureC = futureA.applyToEither(futureB, product -> "结果:" + product);
    //结果:通过方式A获取商品a
    System.out.println(futureC.get());
}

（2）acceptEither

（3）runAfterEither

5. 级联任务

方法	有无返回值	描述
thenCompose	有	当原任务完成后，以其结果为参数，返回一个新的任务（而不是新结果，类似flatMap）

（1）thenCompose

　　这个方法等待第一个阶段的完成(大写转换)，它的结果传给一个指定的返回CompletableFuture函数，它的结果就是返回的CompletableFuture的结果。

源码

java 复制代码

public <U> CompletableFuture<U> thenCompose(
    Function<? super T, ? extends CompletionStage<U>> fn) {
    return uniComposeStage(null, fn);
}

示例

java 复制代码

private static void thenCompose() {
    String original = "Message";
    CompletableFuture cf = CompletableFuture.completedFuture(original).thenApply(s -> delayedUpperCase(s))
            .thenCompose(upper -> CompletableFuture.completedFuture(original).thenApply(s -> delayedLowerCase(s))
                    .thenApply(s -> upper + s));
    // MESSAGEmessage
    System.out.println(cf.join());
}

6. 单任务结果或异常消费

方法	有无返回值	描述
handle	有	任务完成后执行BiFunction，结果转换，入参为结果或者异常，返回新结果
whenComplete	无	任务完成后执行BiConsumer，结果消费，入参为结果或者异常，不返回新结果
exceptionally	无	任务异常，则执行Function，异常转换，入参为原任务的异常信息，若原任务无异常，则返回原任务结果，即不执行转换

异常流程

java 复制代码

CompletableFuture.supplyAsync(() -> "resultA")
              .thenApply(resultA -> resultA + " resultB")
              .thenApply(resultB -> resultB + " resultC")
              .thenApply(resultC -> resultC + " resultD");

　　上面的代码中，任务 A、B、C、D 依次执行，如果任务 A 抛出异常（当然上面的代码不会抛出异常），那么后面的任务都得不到执行。如果任务 C 抛出异常，那么任务 D 得不到执行。

　　那么我们怎么处理异常呢？看下面的代码，我们在任务 A 中抛出异常，并对其进行处理：

（1）handle

示例

java 复制代码

private static void handle() {
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "resultA")
            .thenApply(resultA -> resultA + " resultB")
            // 任务 C 抛出异常
            .thenApply(resultB -> {throw new RuntimeException();})
            // 处理任务 C 的返回值或异常
            .handle(new BiFunction<Object, Throwable, Object>() {
                @Override
                public Object apply(Object re, Throwable throwable) {
                    if (throwable != null) {
                        return "errorResultC";
                    }
                    return re;
                }
            })
            .thenApply(resultC -> {
                System.out.println("resultC:" + resultC);
                return resultC + " resultD";
            });
    System.out.println(future.join());
}

（2）whenComplete

示例

java 复制代码

private static void whenComplete() throws ExecutionException, InterruptedException {
    // 创建异步执行任务:
    CompletableFuture<Double> cf = CompletableFuture.supplyAsync(()->{
        System.out.println(Thread.currentThread()+"job1 start,time->"+System.currentTimeMillis());
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
        }
        if(true){
            throw new RuntimeException("test");
        }else{
            System.out.println(Thread.currentThread()+"job1 exit,time->"+System.currentTimeMillis());
            return 1.2;
        }
    });
    //cf执行完成后会将执行结果和执行过程中抛出的异常传入回调方法
    // 如果是正常执行，a=1.2，b则传入的异常为null
    //如果异常执行，a=null，b则传入异常信息
    CompletableFuture<Double> cf2=cf.whenComplete((a,b)->{
        System.out.println(Thread.currentThread()+"job2 start,time->"+System.currentTimeMillis());
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
        }
        if(b!=null){
            System.out.println("error stack trace->");
            b.printStackTrace();
        }else{
            System.out.println("run succ,result->"+a);
        }
        System.out.println(Thread.currentThread()+"job2 exit,time->"+System.currentTimeMillis());
    });
    //等待子任务执行完成
    System.out.println("main thread start wait,time->"+System.currentTimeMillis());
    //如果cf是正常执行的，cf2.get的结果就是cf执行的结果
    //如果cf是执行异常，则cf2.get会抛出异常
    System.out.println("run result->"+cf2.get());
    System.out.println("main thread exit,time->"+System.currentTimeMillis());
}

（3）exceptionally

7. 合并多个complete为一个

方法		描述
allOf		合并多个complete为一个，等待全部完成
anyOf		合并多个complete为一个，等待其中之一完成

（1）allOf

　　我们在处理业务时，有时会有多任务异步处理，同步返回结果的情况

采用多线程执异步行某种任务，比如在不同主机查询磁盘列表信息。
将执行结果搜集，分组分类，处理。
将处理以后的结果给予展示。

示例

java 复制代码

 // 创建异步执行任务:
        CompletableFuture<Double> cf = CompletableFuture.supplyAsync(()->{
            System.out.println(Thread.currentThread()+" start job1,time->"+System.currentTimeMillis());
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
            }
            System.out.println(Thread.currentThread()+" exit job1,time->"+System.currentTimeMillis());
            return 1.2;
        });
        CompletableFuture<Double> cf2 = CompletableFuture.supplyAsync(()->{
            System.out.println(Thread.currentThread()+" start job2,time->"+System.currentTimeMillis());
            try {
                Thread.sleep(1500);
            } catch (InterruptedException e) {
            }
            System.out.println(Thread.currentThread()+" exit job2,time->"+System.currentTimeMillis());
            return 3.2;
        });
        CompletableFuture<Double> cf3 = CompletableFuture.supplyAsync(()->{
            System.out.println(Thread.currentThread()+" start job3,time->"+System.currentTimeMillis());
            try {
                Thread.sleep(1300);
            } catch (InterruptedException e) {
            }
//            throw new RuntimeException("test");
            System.out.println(Thread.currentThread()+" exit job3,time->"+System.currentTimeMillis());
            return 2.2;
        });
        //allof等待所有任务执行完成才执行cf4，如果有一个任务异常终止，则cf4.get时会抛出异常，都是正常执行，cf4.get返回null
        //anyOf是只有一个任务执行完成，无论是正常执行或者执行异常，都会执行cf4，cf4.get的结果就是已执行完成的任务的执行结果
        CompletableFuture cf4=CompletableFuture.allOf(cf,cf2,cf3).whenComplete((a,b)->{
            if(b!=null){
                System.out.println("error stack trace->");
                b.printStackTrace();
            }else{
                System.out.println("run succ,result->"+a);
            }
        });
        System.out.println("main thread start cf4.get(),time->"+System.currentTimeMillis());
        //等待子任务执行完成
        System.out.println("cf4 run result->"+cf4.get());
        System.out.println("main thread exit,time->"+System.currentTimeMillis());

获取返回值方法

java 复制代码

public <T> CompletableFuture<List<T>> allOf(List<CompletableFuture<T>> futuresList) {
    CompletableFuture<Void> allFuturesResult =
            CompletableFuture.allOf(futuresList.toArray(new CompletableFuture[futuresList.size()]));
    return allFuturesResult.thenApply(v ->
            futuresList.stream().
                    map(future -> future.join()).
                    collect(Collectors.<T>toList())
    );
}

（2）anyOf

　　CompletableFuture.anyOf()和其名字介绍的一样，当任何一个CompletableFuture完成的时候【相同的结果类型】，返回一个新的CompletableFuture。

　　示例

java 复制代码

private static void anyOf() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                throw new IllegalStateException(e);
            }
            return "Result of Future 1";
        });

        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                throw new IllegalStateException(e);
            }
            return "Result of Future 2";
        });

        CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                throw new IllegalStateException(e);
            }
            return "Result of Future 3";
        });
        CompletableFuture<Object> anyOfFuture = CompletableFuture.anyOf(future1, future2, future3);
        System.out.println(anyOfFuture.get()); // Result of Future 2
    }

三、其他相关api

1. future接口

（1）isDone()

　　判断任务是否完成。三种完成情况：normally（正常执行完毕）、exceptionally（执行异常）、via cancellation（取消）

（2）get()

　　阻塞获取结果或抛出受检测异常，需要显示进行try...catch处理。

（3）get(long timeout,TimeUnit unit)

　　超时阻塞获取结果

（4）cancel(boolean mayInterruptIfRunning)

　　取消任务，若一个任务未完成，则以CancellationException异常。其相关未完成的子任务也会以CompletionException结束

（5）isCancelled()

　　是否已取消，在任务正常执行完成前取消，才为true。否则为false。

2. CompletableFuture接口

（1）join

阻塞获取结果或抛出非受检异常。

（2）getNow(T valueIfAbsent)

若当前任务无结果，则返回valueIfAbsent，否则返回已完成任务的结果。

（3）complete(T value)

设置任务结果，任务正常结束，之后的任务状态为已完成。

（4）completeExceptionally(Throwable ex)

设置任务异常结果，任务异常结束，之后的任务状态为已完成。

（5）isCompletedExceptionally()

判断任务是否异常结束。异常可能的原因有：取消、显示设置任务异常结果、任务动作执行异常等。

（6）getNumberOfDependents()

返回依赖当前任务的任务数量，主要用于监控。

（7）orTimeout(long timeout,TimeUnit unit) jdk9

设置任务完成超时时间，若在指定时间内未正常完成，则任务会以异常(TimeoutException)结束。

（8）completeOnTimeout(T value,long timeout,TimeUnit unit) jdk9

设置任务完成超时时间，若在指定时间内未正常完成，则以给定的value为任务结果

四、实战

1. API网关做接口的聚合

java 复制代码

//这两个参数从外部获得
Long userId = 10006L;
String orderId = "XXXXXXXXXXXXXXXXXXXXXX";
//从用户服务获取用户信息
UserInfo userInfo = userService.getUserInfo(userId);
//从用订单务获取订单信息
OrderInfo orderInfo = orderService.getOrderInfo(orderId);
//返回两者的聚合DTO
return new OrderDetailDTO(userInfo,orderInfo);

下面三个外部接口的信息一定是不相关联的，也就是可以并行获取，三个接口的结果都获取完毕之后做一次数据聚合到DTO即可，也就是聚合的耗时大致是这三个接口中耗时最长的接口的响应时间

java 复制代码

@Service
public class OrderDetailService {
	/**
	 * 建立一个线程池专门交给CompletableFuture使用
	 */
	private final ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 20, 0, TimeUnit.SECONDS,
			new ArrayBlockingQueue<>(100));
	@Autowired
	private UserService userService;
	@Autowired
	private OrderService orderService;
	public OrderDetailDTO getOrderDetail(Long userId, String orderId) throws Exception {
		CompletableFuture<UserInfo> userInfoCompletableFuture = CompletableFuture.supplyAsync(() -> userService.getUserInfo(userId), executor);
		CompletableFuture<OrderInfo> orderInfoCompletableFuture = CompletableFuture.supplyAsync(() -> orderService.getOrderInfo(orderId), executor);
		CompletableFuture<OrderDetailDTO> result
				= userInfoCompletableFuture.thenCombineAsync(orderInfoCompletableFuture, OrderDetailDTO::new, executor);
		return result.get();
	}
}

五、区别

（1）whenComplete和handle区别

　　whenComplete 与 handle 方法就类似于 try..catch..finanlly 中 finally 代码块。无论是否发生异常，都将会执行的。这两个方法区别在于 handle 支持返回结果。

（2）thenApply与thenCompose的异同

对于thenApply，fn函数是一个对一个已完成的stage或者说CompletableFuture的的返回值进行计算、操作；

对于thenCompose，fn函数是对另一个CompletableFuture进行计算、操作。

（3）有无Async的区别

　　没有Async的在CompleteableFuture调用它的线程定义的线程上运行，因此通常不知道在哪里执行该线程。如果结果已经可用，它可能会立即执行。

有Async的无论环境如何，都在环境定义的执行程序上运行。为此CompletableFuture通常ForkJoinPool.commonPool()。

www.apiref.com/java11-zh/j…

colobu.com/2018/03/12/…

欢迎大家关注微信公众号：CodingTao

标签：return,System,任务,详解,CompletableFuture,println,out
From： https://www.cnblogs.com/tiancai/p/17547428.html

CompletableFuture详解

CompletableFuture

前言

一、创建方式

1. 用默认线程池

2. 用自定义线程池

二、使用示例

1. 构建异步任务

（1）runAsync

（2）supplyAsync

2. 单任务结果消费

（1）thenApply

（2）thenAccept

（3）thenRun

3. 合并结果消费

（1）thenCombine

（2）thenAcceptBoth

（3）runAfterBoth

4. 任一结果消费

（1）applyToEither

（2）acceptEither

（3）runAfterEither

5. 级联任务

（1）thenCompose

6. 单任务结果或异常消费

（1）handle

（2）whenComplete

（3）exceptionally

7. 合并多个complete为一个

（1）allOf

（2）anyOf

三、其他相关api

1. future接口

（1）isDone()

（2）get()

（3）get(long timeout,TimeUnit unit)

（4）cancel(boolean mayInterruptIfRunning)

（5）isCancelled()

2. CompletableFuture接口

（1）join

（2）getNow(T valueIfAbsent)

（3）complete(T value)

（4）completeExceptionally(Throwable ex)

（5）isCompletedExceptionally()

（6）getNumberOfDependents()

（7）orTimeout(long timeout,TimeUnit unit) jdk9

（8）completeOnTimeout(T value,long timeout,TimeUnit unit) jdk9

四、实战

1. API网关做接口的聚合

五、区别

（1）whenComplete和handle区别

（2）thenApply与thenCompose的异同

（3）有无Async的区别

相关文章

赞助商

阅读排行