1.创建异步对象
CompletableFuture提供四个静态方法来创建一个异步操作。
//runAsync，传入Runnable ，返回CompletableFuture<Void> ，无返回值，Executor 可以指定一个线程池
public static CompletableFuture<Void> runAsync(Runnable runnable)
public static CompletableFuture<Void> runAsync(Runnable runnable,Executor executor)
//supplyAsync，有返回值
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,Executor executor)
runAsync举例：
public class completable {
//创建一个线程池
public static ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());
public static void main(String[] args) {
System.out.println("main start");
CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
System.out.println("当前线程：" + Thread.currentThread().getName());
int i = 10 / 2;
System.out.println("运行结果：" + i);
}, executor);
System.out.println("main end");
}
}
main start
main end
当前线程：pool-1-thread-1
运行结果：5
supplyAsync举例：
pblic class completable {
//创建一个线程池
public static ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());
public static void main(String[] args) throws ExecutionException, InterruptedException {
System.out.println("main start");
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
System.out.println("当前线程：" + Thread.currentThread().getName());
int i = 10 / 2;
System.out.println("运行结果：" + i);
return i;
}, executor);
Integer result = future.get();
System.out.println("main end,返回结果：" + result);
}
}
main start
当前线程：pool-1-thread-1
运行结果：5
main end,返回结果：5
2.计算完成时回调方法
//上一个任务完成以后，再用同一个线程去调用下一个任务
public CompletableFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action)
//上一个任务完成以后，再用异步的方式去调用下一个任务
public CompletionStage<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action);
public CompletionStage<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor);
//处理异常情况
public CompletableFuture<T> exceptionally(Function<Throwable, ? extends T> fn)
whenComplete例子（无异常）：
public class completable {
//创建一个线程池
public static ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());
public static void main(String[] args) throws ExecutionException, InterruptedException {
System.out.println("main start");
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
System.out.println("当前线程：" + Thread.currentThread().getName());
int i = 10 /0;
System.out.println("运行结果：" + i);
return i;
}, executor).whenComplete((res, exception) -> {
System.out.println("异步任务成功完成了。。结果是：" + res + ",异常是：" + exception);
});
System.out.println("main end");
}
}
main start
main end
当前线程：pool-1-thread-1
运行结果：2
异步任务成功完成了。。结果是：2,异常是：null
whenComplete例子（有异常）：
public class completable {
//创建一个线程池
public static ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());
public static void main(String[] args) throws ExecutionException, InterruptedException {
System.out.println("main start");
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
System.out.println("当前线程：" + Thread.currentThread().getName());
int i = 10 / 0;
System.out.println("运行结果：" + i);
return i;
}, executor).whenComplete((res, exception) -> {
System.out.println("异步任务成功完成了。。结果是：" + res + ",异常是：" + exception);
});
System.out.println("main end");
}
}
main start
main end
当前线程：pool-1-thread-1
异步任务成功完成了。。结果是：null,异常是：java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero
exceptionally例子：
public class completable {
//创建一个线程池
public static ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());
public static void main(String[] args) throws ExecutionException, InterruptedException {
System.out.println("main start");
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
System.out.println("当前线程：" + Thread.currentThread().getName());
int i = 10 / 0;
System.out.println("运行结果：" + i);
return i;
}, executor).whenComplete((res, exception) -> {
// 虽然可以得到异常信息，但是无法修改返回数据
System.out.println("异步任务成功完成了。。结果是：" + res + ",异常是：" + exception);
}).exceptionally(throwable -> {
// 可以处理异常，同时在异常出现后返回一个值
return 10;
});
System.out.println("main end,返回结果："+future.get());
}
}
main start
当前线程：pool-1-thread-1
异步任务成功完成了。。结果是：null,异常是：java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero
main end,返回结果：10
3.handle方法
和complete一样，可以对结果做最后的处理（可以处理异常），可以改变返回值
public <U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn)
public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn)
public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor)
例子
public class completable {
//创建一个线程池
public static ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());
public static void main(String[] args) throws ExecutionException, InterruptedException {
System.out.println("main start");
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
System.out.println("当前线程：" + Thread.currentThread().getName());
int i = 10 / 0;
System.out.println("运行结果：" + i);
return i;
}, executor).handle((res, exception) -> {
if (res != null) {
return res * 2;
}
if (exception != null) {
return 0;
}
return 0;
});
System.out.println("main end,返回结果：" + future.get());
}
}
main start
当前线程：pool-1-thread-1
main end,返回结果：0
4.线程串行化方法
//thenApply：当一个线程依赖另一个线程时，获取上一个任务返回的结果 ，并返回当前任务的返回值。
//带有Async默认是异步执行任务。可指定线程池
public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn)
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)
//thenAccept：消费处理结果 。接收任务的处理结果 ，并消费处理，无返回结果
public CompletableFuture<Void> thenAccept(Consumer<? super T> action)
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action)
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor)
//thenRun：只要上面的任务执行完成，就开始执行thenRun的后续操作。
public CompletableFuture<Void> thenRun(Runnable action)
public CompletableFuture<Void> thenRunAsync(Runnable action)
public CompletableFuture<Void> thenRunAsync(Runnable action,Executor executor)
举例
public class completable {
//创建一个线程池
public static ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());
public static void main(String[] args) {
System.out.println("main start");
/*
* thenRunAsync无法获取上一步执行结果
* 如果上一任务抛出异常，thenRunAsync也不会执行
*/
CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
System.out.println("当前线程：" + Thread.currentThread().getName());
int i = 10 / 4;
System.out.println("运行结果：" + i);
return i;
}, executor).thenRunAsync(() -> {
System.out.println("任务2启动了");
}, executor);
System.out.println("main end");
}
}
main start
main end
当前线程：pool-1-thread-1
public class completable {
//创建一个线程池
public static ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());
public static void main(String[] args) {
System.out.println("main start");
/*
* thenAccept可以获取上一步返回结果
*/
CompletableFuture.supplyAsync(() -> {
System.out.println("当前线程：" + Thread.currentThread().getName());
int i = 10 / 4;
System.out.println("运行结果：" + i);
return i;
}, executor).thenAccept(res -> {
System.out.println("当前线程：" + Thread.currentThread().getName());
System.out.println("任务2启动了，上一任务执行结果：" + res);
});
System.out.println("main end");
}
}
main start
main end
当前线程：pool-1-thread-1
运行结果：2
当前线程：pool-1-thread-1
任务2启动了，上一任务执行结果：2
public class completable {
//创建一个线程池
public static ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());
public static void main(String[] args) throws ExecutionException, InterruptedException {
System.out.println("main start");
/*
* thenApplyAsync 带有返回值，可以返回结果
*/
CompletableFuture<Object> future = CompletableFuture.supplyAsync(() -> {
System.out.println("当前线程：" + Thread.currentThread().getName());
int i = 10 / 4;
System.out.println("运行结果：" + i);
return i;
}, executor).thenApplyAsync(res -> {
System.out.println("任务2启动了,上一步结果：" + res);
return "hello " + res;
}, executor);
System.out.println("main end,最终返回结果：" + future.get());
}
}
main start
当前线程：pool-1-thread-1
运行结果：2
任务2启动了,上一步结果：2
main end,最终返回结果：hello 2
5.两任务组合——任务都要完成
// thenCombine：组合两个future,获取两个future的返回结果，并返回当前任务的返回值
public <U,V> CompletableFuture<V> thenCombine(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn)
public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn)
public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn, Executor executor)
// thenAcceptBoth：组合两个future，获取两个future任务的返回结果 ，然后处理任务，没有返回值。
public <U> CompletableFuture<Void> thenAcceptBoth(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action)
public <U> CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action)
public <U> CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action, Executor executor)
// runAfterBoth：组合两个future，不需要获取future的结果 ，只需要两个future处理完任务后，处理该任务。
public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other, Runnable action)
public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,Runnable action)
public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,Runnable action,Executor executor)
例子
public class completable {
//创建一个线程池
public static ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());
public static void main(String[] args) throws ExecutionException, InterruptedException {
System.out.println("main start");
CompletableFuture<Integer> future01 = CompletableFuture.supplyAsync(() -> {
System.out.println("任务1线程：" + Thread.currentThread().getName());
int i = 10 / 4;
System.out.println("任务1结束，结果：" + i);
return i;
}, executor);
CompletableFuture<String> future02 = CompletableFuture.supplyAsync(() -> {
System.out.println("任务2线程：" + Thread.currentThread().getName());
System.out.println("任务2结束");
return "hello";
}, executor);
//任务3会等任务1和任务2结束以后再执行。
future01.runAfterBothAsync(future02,()-> System.out.println("任务3开始。"),executor);
System.out.println("main end");
}
}
main start
main end
任务1线程：pool-1-thread-1
任务1结束，结果：2
任务2线程：pool-1-thread-2
任务2结束
任务3开始。
public class completable {
//创建一个线程池
public static ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());
public static void main(String[] args) throws ExecutionException, InterruptedException {
System.out.println("main start");
CompletableFuture<Integer> future01 = CompletableFuture.supplyAsync(() -> {
System.out.println("任务1线程：" + Thread.currentThread().getName());
int i = 10 / 4;
System.out.println("任务1结束，结果：" + i);
return i;
}, executor);
CompletableFuture<String> future02 = CompletableFuture.supplyAsync(() -> {
System.out.println("任务2线程：" + Thread.currentThread().getName());
System.out.println("任务2结束");
return "hello";
}, executor);
// future01.runAfterBothAsync(future02,()-> System.out.println("任务3开始。"),executor);
future01.thenAcceptBothAsync(future02, (f1, f2) -> System.out.println("任务3开始。任务1结果：" + f1 + " 任务2结果：" + f2), executor);
System.out.println("main end");
}
}
main start
main end
任务1线程：pool-1-thread-1
任务1结束，结果：2
任务2线程：pool-1-thread-2
任务2结束
任务3开始。任务1结果：2 任务2结果：hello
public class completable {
//创建一个线程池
public static ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());
public static void main(String[] args) throws ExecutionException, InterruptedException {
System.out.println("main start");
CompletableFuture<Integer> future01 = CompletableFuture.supplyAsync(() -> {
System.out.println("任务1线程：" + Thread.currentThread().getName());
int i = 10 / 4;
System.out.println("任务1结束，结果：" + i);
return i;
}, executor);
CompletableFuture<String> future02 = CompletableFuture.supplyAsync(() -> {
System.out.println("任务2线程：" + Thread.currentThread().getName());
System.out.println("任务2结束");
return "hello";
}, executor);
// future01.runAfterBothAsync(future02,()-> System.out.println("任务3开始。"),executor);
// future01.thenAcceptBothAsync(future02, (f1, f2) -> System.out.println("任务3开始。任务1结果：" + f1 + " 任务2结果：" + f2), executor);
CompletableFuture<String> future = future01.thenCombineAsync(future02, (f1, f2) -> f1 + ":" + f2 + "->haha", executor);
System.out.println("main end,最终返回结果:"+future.get());
}
}
main start
任务1线程：pool-1-thread-1
任务1结束，结果：2
任务2线程：pool-1-thread-2
任务2结束
main end,最终返回结果:2:hello->haha
6.两任务组合——任意一个任务完成
// applyToEither: 两个任务有一个执行完成，获取它的返回值。处理任务并有新的返回值。
public <U> CompletableFuture<U> applyToEither(CompletionStage<? extends T> other, Function<? super T, U> fn)
public <U> CompletableFuture<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn)
public <U> CompletableFuture<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn,Executor executor)
// acceptEither：两个任务有一个执行完成,获取它的返回值，处理任务，没有新的返回值。
public CompletableFuture<Void> acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action)
public CompletableFuture<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action)
public CompletableFuture<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action,Executor executor)
// runAfterEither：两个任务有一个执行完成，不需要获取返回值，处理第三个任务，也没有返回值。
public CompletableFuture<Void> runAfterEither(CompletionStage<?> other,Runnable action)
public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other,Runnable action)
public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other,Runnable action,Executor etxecutor)
7.多任务组合
// allOf：等待所有任务完成
public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)
// anyOf：只要有一个任务完成
public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs)
