线程异步编排并行（CompletableFuture）

Posted 2022-12-02 csdn_20210509

美图

两个任务组合（一）

任务一和任务二都处理完成之后，才能处理任务三

runAfterBoth

组合两个future，不需要获取future的结果，只需要两个future处理完成后，处理该任务，无返回结果

CompletableFuture<Void> runAfterBoth(CompletionStage<?> other, Runnable action)

CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action)

CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action, Executor executor)

示例

CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> System.out.println("任务1"));
CompletableFuture<Void> future2 = CompletableFuture.runAsync(() -> System.out.println("任务2"));
future1.runAfterBoth(future2, () -> System.out.println("任务3"));

thenAcceptBoth

组合两个future，需要获取future的结果，只需要两个future处理完成后，处理该任务，无返回结果

CompletableFuture<Void> thenAcceptBoth(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action)

CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action)

CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action, Executor executor)

示例

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 
    System.out.println("任务1");
    return 10;
);
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 
    System.out.println("任务2");
    return 20;
);
future1.thenAcceptBoth(future2, (res1, res2) -> 
    System.out.println("任务3");
    System.out.println(res1+res2);
);

thenCombine

组合两个future，需要获取future的结果，只需要两个future处理完成后，处理该任务，有返回结果

CompletableFuture<V> thenCombine(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)

CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)

CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn, Executor executor)

示例

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 
    System.out.println("任务1");
    return 10;
);
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 
    System.out.println("任务2");
    return 20;
);
Integer integer = future1.thenCombine(future2, (res1, res2) -> 
    System.out.println("任务3");
    return res1 + res2;
).get();
System.out.println(integer);

两个任务组合（二）

任务一和任务二任意一个处理完成之后，处理任务三

runAfterEither

组合两个future，不需要获取future的结果，任意一个future处理完成后，处理该任务，无返回结果

CompletableFuture<Void> runAfterEither(CompletionStage<?> other, Runnable action)

CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action)

CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action, Executor executor)

示例

CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> 
    try 
        Thread.sleep(1000);
     catch (InterruptedException e) 
        e.printStackTrace();
    
    System.out.println("任务1");
);
CompletableFuture<Void> future2 = CompletableFuture.runAsync(() -> 
    try 
        Thread.sleep(3000);
     catch (InterruptedException e) 
        e.printStackTrace();
    
    System.out.println("任务2");
);

future1.runAfterEither(future2, () -> System.out.println("任务3"));

Thread.sleep(4000);

acceptEither

组合两个future，需要获取future的结果，只需要两个future处理完成后，处理该任务，无返回结果

CompletableFuture<Void> acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action)

CompletableFuture<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action)

CompletableFuture<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action, Executor executor)

示例

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 
    try 
        Thread.sleep(1000);
     catch (InterruptedException e) 
        e.printStackTrace();
    
    System.out.println("任务1");
    return 10;
);
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 
    try 
        Thread.sleep(3000);
     catch (InterruptedException e) 
        e.printStackTrace();
    
    System.out.println("任务2");
    return 20;
);

future1.acceptEither(future2, (res) -> 
    System.out.println("任务3");
    System.out.println(res);
);

Thread.sleep(4000);

applyToEither

组合两个future，需要获取future的结果，只需要两个future处理完成后，处理该任务，有返回结果

CompletableFuture<U> applyToEither(CompletionStage<? extends T> other, Function<? super T, U> fn)

CompletableFuture<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn)

CompletableFuture<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn, Executor executor)

示例

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 
    try 
        Thread.sleep(1000);
     catch (InterruptedException e) 
        e.printStackTrace();
    
    System.out.println("任务1");
    return 10;
);
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 
    try 
        Thread.sleep(3000);
     catch (InterruptedException e) 
        e.printStackTrace();
    
    System.out.println("任务2");
    return 20;
);

Integer integer = future1.applyToEither(future2, (res) -> 
    System.out.println("任务3");
    return res;
).get();

Thread.sleep(4000);

多任务组合

allOf

需要所有的都执行完成

CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)

示例

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 
    try 
        Thread.sleep(1000);
     catch (InterruptedException e) 
        e.printStackTrace();
    
    System.out.println("任务1");
    return 10;
);
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 
    try 
        Thread.sleep(3000);
     catch (InterruptedException e) 
        e.printStackTrace();
    
    System.out.println("任务2");
    return 20;
);

CompletableFuture.allOf(future1, future2).get();
Thread.sleep(4000);

CompletableFuture.allOf(future1, future2).get(); 必须调用get(),主线程才会在此等待

anyOf

需要任意一个执行完成

CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs)

示例

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 
    try 
        Thread.sleep(1000);
     catch (InterruptedException e) 
        e.printStackTrace();
    
    System.out.println("任务1");
    return 10;
);
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 
    try 
        Thread.sleep(3000);
     catch (InterruptedException e) 
        e.printStackTrace();
    
    System.out.println("任务2");
    return 20;
);

Object o = CompletableFuture.anyOf(future1, future2).get();
Thread.sleep(4000);