晋中路桥建设集团网站,企业网站开发的公司,网络营销工具的特点,郑州网站建设外包业务JAVA重要知识点CompletableFuture常见函数式编程操作创建 CompletableFuture静态工厂方法处理异步结算的结果异常处理组合 CompletableFuturethenCompose() 和 thenCombine() 区别并行运行多个 CompletableFutureCompletableFuture
Java 8 才被引入的一个非常有用的用于异步编…
JAVA重要知识点CompletableFuture常见函数式编程操作创建 CompletableFuture静态工厂方法处理异步结算的结果异常处理组合 CompletableFuturethenCompose() 和 thenCombine() 区别并行运行多个 CompletableFutureCompletableFuture
Java 8 才被引入的一个非常有用的用于异步编程的类。 CompletableFuture 同时实现了 Future 和 CompletionStage 接口。
public class CompletableFutureT implements FutureT,
CompletionStageT {
}CompletableFuture 除了提供了更为好用和强大的 Future 特性之外还提供了函数式编程的能力。 Future 接口有 5 个方法 boolean cancel(boolean mayInterruptIfRunning) 尝试取消执行任务。 boolean isCancelled() 判断任务是否被取消。 boolean isDone() 判断任务是否已经被执行完成。 get() 等待任务执行完成并获取运算结果。 get(long timeout, TimeUnit unit) 多了一个超时时间。
CompletionStage 接口中的方法比较多CompletableFuture 的函数式能力就是这个接口赋予的。从这个接口的方法参数你就可以发现其大量使用了 Java8 引入的函数式编程。
常见函数式编程操作
创建 CompletableFuture
常见的创建 CompletableFuture 对象的方法如下
通过 new 关键字。基于 CompletableFuture 自带的静态工厂方法runAsync() 、supplyAsync() 。
new关键字 通过 new 关键字创建 CompletableFuture 对象这种使用方式可以看作是将 CompletableFuture 当做 Future 来使用。 举例 创建一个结果值类型为 RpcResponse Object 的 CompletableFuture可以把 resultFuture 看作是异步运算结果的载体。
CompletableFutureRpcResponseObject resultFuture new CompletableFuture();调用 complete() 方法为其传入结果这表示 resultFuture 已经被完成了
// complete() 方法只能调用一次后续调用将被忽略。
resultFuture.complete(rpcResponse);可以通过 isDone() 方法来检查是否已经完成
public boolean isDone() {return result ! null;
}获取异步计算的结果也非常简单直接调用 get() 方法即可。调用 get() 方法的线程会阻塞直到 CompletableFuture 完成运算。
rpcResponse completableFuture.get();如果已经知道计算的结果的话可以使用静态方法 completedFuture() 来创建 CompletableFuture
CompletableFutureString future
CompletableFuture.completedFuture(hello!);
assertEquals(hello!, future.get());
completedFuture() 方法底层调用的是带参数的 new 方法只不过这个方法不对外暴露
public static U CompletableFutureU completedFuture(U value) {return new CompletableFutureU((value null) ? NIL : value);
}
静态工厂方法
supplyAsync和runAsync可以帮助封装计算逻辑
static U CompletableFutureU supplyAsync(SupplierU supplier);
// 使用自定义线程池(推荐)
static U CompletableFutureU supplyAsync(SupplierU supplier, Executor executor);
static CompletableFutureVoid runAsync(Runnable runnable);
// 使用自定义线程池(推荐)
static CompletableFutureVoid runAsync(Runnable runnable, Executor executor);
runAsync() 方法接受的参数是 Runnable 这是一个函数式接口不允许返回值。当你需要异步操作且不关心返回结果的时候可以使用 runAsync() 方法。
FunctionalInterface
public interface Runnable {public abstract void run();
}
supplyAsync() 方法接受的参数是 Supplier U 这也是一个函数式接口U 是返回结果值的类型。
FunctionalInterface
public interface SupplierT {/*** Gets a result.** return a result*/T get();
}需要异步操作且关心返回结果的时候,可以使用 supplyAsync() 方法。举例如下
CompletableFutureVoid future CompletableFuture.runAsync(() - System.out.println(hello!));
future.get();// 输出 hello!
CompletableFutureString future2 CompletableFuture.supplyAsync(() - hello!);
assertEquals(hello!, future2.get());
处理异步结算的结果
获取到异步计算的结果之后还可以对其进行进一步的处理比较常用的方法有下面几个 thenApply() thenAccept() thenRun() whenComplete()
thenApply() 方法接受一个 Function 实例用它来处理结果。方法源码如下
// 沿用上一个任务的线程池
public U CompletableFutureU thenApply(Function? super T,? extends U fn) {return uniApplyStage(null, fn);
}//使用默认的 ForkJoinPool 线程池不推荐
public U CompletableFutureU thenApplyAsync(Function? super T,? extends U fn) {return uniApplyStage(defaultExecutor(), fn);
}
// 使用自定义线程池(推荐)
public U CompletableFutureU thenApplyAsync(Function? super T,? extends U fn, Executor executor) {return uniApplyStage(screenExecutor(executor), fn);
}thenApply() 方法使用示例如下
CompletableFutureString future CompletableFuture.completedFuture(hello!).thenApply(s - s world!);
assertEquals(hello!world!, future.get());
// 这次调用将被忽略。
future.thenApply(s - s nice!);
assertEquals(hello!world!, future.get());流式调用如下
CompletableFutureString future CompletableFuture.completedFuture(hello!).thenApply(s - s world!).thenApply(s - s nice!);
assertEquals(hello!world!nice!, future.get());
如果不需要从回调函数中获取返回结果可以使用 thenAccept() 或者 thenRun()。这两个方法的区别在于 thenRun() 不能访问异步计算的结果。
thenAccept() 方法的参数是 Consumer? super T 。源码如下
public CompletableFutureVoid thenAccept(Consumer? super T action) {return uniAcceptStage(null, action);
}public CompletableFutureVoid thenAcceptAsync(Consumer? super T action) {return uniAcceptStage(defaultExecutor(), action);
}public CompletableFutureVoid thenAcceptAsync(Consumer? super T action,Executor executor) {return uniAcceptStage(screenExecutor(executor), action);
}
解释上述代码Consumer 属于消费型接口它可以接收 1 个输入对象然后进行“消费”。接口源码如下
FunctionalInterface
public interface ConsumerT {void accept(T t);default ConsumerT andThen(Consumer? super T after) {Objects.requireNonNull(after);return (T t) - { accept(t); after.accept(t); };}
}
thenRun() 的方法是的参数是 Runnable。如下
public CompletableFutureVoid thenRun(Runnable action) {return uniRunStage(null, action);
}public CompletableFutureVoid thenRunAsync(Runnable action) {return uniRunStage(defaultExecutor(), action);
}public CompletableFutureVoid thenRunAsync(Runnable action,Executor executor) {return uniRunStage(screenExecutor(executor), action);
}thenAccept() 和 thenRun() 使用示例如下
CompletableFuture.completedFuture(hello!).thenApply(s - s world!).thenApply(s - s nice!).thenAccept(System.out::println);//hello!world!nice!CompletableFuture.completedFuture(hello!).thenApply(s - s world!).thenApply(s - s nice!).thenRun(() - System.out.println(hello!));//hello!
上述两个方法结合起来后使用举例如下
CompletableFuture.completedFuture(hello!).thenApply(s - s world!).thenApply(s - s nice!).thenAccept(System.out::println);//hello!world!nice!CompletableFuture.completedFuture(hello!).thenApply(s - s world!).thenApply(s - s nice!).thenRun(() - System.out.println(hello!));//hello!
whenComplete() 的方法的参数是 BiConsumer? super T, ? super Throwable 。
public CompletableFutureT whenComplete(BiConsumer? super T, ? super Throwable action) {return uniWhenCompleteStage(null, action);
}public CompletableFutureT whenCompleteAsync(BiConsumer? super T, ? super Throwable action) {return uniWhenCompleteStage(defaultExecutor(), action);
}
// 使用自定义线程池(推荐)
public CompletableFutureT whenCompleteAsync(BiConsumer? super T, ? super Throwable action, Executor executor) {return uniWhenCompleteStage(screenExecutor(executor), action);
}
BiConsumer 可以接收 2 个输入对象然后进行“消费”
FunctionalInterface
public interface BiConsumerT, U {void accept(T t, U u);default BiConsumerT, U andThen(BiConsumer? super T, ? super U after) {Objects.requireNonNull(after);return (l, r) - {accept(l, r);after.accept(l, r);};}
}
whenComplete() 使用示例如下
CompletableFutureString future CompletableFuture.supplyAsync(() - hello!).whenComplete((res, ex) - {// res 代表返回的结果// ex 的类型为 Throwable 代表抛出的异常System.out.println(res);// 这里没有抛出异常所有为 nullassertNull(ex);});
assertEquals(hello!, future.get());异常处理
通过 handle() 方法来处理任务执行过程中可能出现的抛出异常的情况
public U CompletableFutureU handle(BiFunction? super T, Throwable, ? extends U fn) {return uniHandleStage(null, fn);
}public U CompletableFutureU handleAsync(BiFunction? super T, Throwable, ? extends U fn) {return uniHandleStage(defaultExecutor(), fn);
}public U CompletableFutureU handleAsync(BiFunction? super T, Throwable, ? extends U fn, Executor executor) {return uniHandleStage(screenExecutor(executor), fn);
}
举例实践如下
CompletableFutureString future CompletableFuture.supplyAsync(() - {if (true) {throw new RuntimeException(Computation error!);}return hello!;
}).handle((res, ex) - {// res 代表返回的结果// ex 的类型为 Throwable 代表抛出的异常return res ! null ? res : world!;
});
assertEquals(world!, future.get());
还可以通过 exceptionally() 方法来处理异常情况
CompletableFutureString future CompletableFuture.supplyAsync(() - {if (true) {throw new RuntimeException(Computation error!);}return hello!;
}).exceptionally(ex - {System.out.println(ex.toString());// CompletionExceptionreturn world!;
});
assertEquals(world!, future.get());可以使用 completeExceptionally() 方法为其赋值令 CompletableFuture 的结果就是异常
CompletableFutureString completableFuture new CompletableFuture();
// ...
completableFuture.completeExceptionally(new RuntimeException(Calculation failed!));
// ...
completableFuture.get(); // ExecutionException
组合 CompletableFuture
可以使用 thenCompose() 按顺序链接两个 CompletableFuture 对象
public U CompletableFutureU thenCompose(Function? super T, ? extends CompletionStageU fn) {return uniComposeStage(null, fn);
}public U CompletableFutureU thenComposeAsync(Function? super T, ? extends CompletionStageU fn) {return uniComposeStage(defaultExecutor(), fn);
}public U CompletableFutureU thenComposeAsync(Function? super T, ? extends CompletionStageU fn,Executor executor) {return uniComposeStage(screenExecutor(executor), fn);
}
thenCompose() 方法使用示例如下
CompletableFutureString future CompletableFuture.supplyAsync(() - hello!).thenCompose(s - CompletableFuture.supplyAsync(() - s world!));
assertEquals(hello!world!, future.get());
类似的还有 thenCombine() 方法 thenCombine() 同样可以组合两个 CompletableFuture 对象
CompletableFutureString completableFuture CompletableFuture.supplyAsync(() - hello!).thenCombine(CompletableFuture.supplyAsync(() - world!), (s1, s2) - s1 s2).thenCompose(s - CompletableFuture.supplyAsync(() - s nice!));
assertEquals(hello!world!nice!, completableFuture.get());
thenCompose() 和 thenCombine() 区别
thenCompose() 可以两个 CompletableFuture 对象并将前一个任务的返回结果作为下一个任务的参数它们之间存在着先后顺序。thenCombine() 会在两个任务都执行完成后把两个任务的结果合并。两个任务是并行执行的它们之间并没有先后依赖顺序。
并行运行多个 CompletableFuture
可以通过 CompletableFuture 的 allOf()这个静态方法来并行运行多个 CompletableFuture 。
实际项目中需要并行运行多个互不相关的任务这些任务之间没有依赖关系可以互相独立地运行。比说要读取处理 6 个文件这 6 个任务都是没有执行顺序依赖的任务但是需要返回给用户的时候将这几个文件的处理的结果进行统计整理。像这种情况我们就可以使用并行运行多个 CompletableFuture 来处理。
CompletableFutureVoid task1 CompletableFuture.supplyAsync(()-{//自定义业务操作});
......
CompletableFutureVoid task6 CompletableFuture.supplyAsync(()-{//自定义业务操作});
......CompletableFutureVoid headerFutureCompletableFuture.allOf(task1,.....,task6);try {headerFuture.join();} catch (Exception ex) {......}
System.out.println(all done. );
allOf() 方法会等到所有的 CompletableFuture 都运行完成之后再返回
Random rand new Random();
CompletableFutureString future1 CompletableFuture.supplyAsync(() - {try {Thread.sleep(1000 rand.nextInt(1000));} catch (InterruptedException e) {e.printStackTrace();} finally {System.out.println(future1 done...);}return abc;
});
CompletableFutureString future2 CompletableFuture.supplyAsync(() - {try {Thread.sleep(1000 rand.nextInt(1000));} catch (InterruptedException e) {e.printStackTrace();} finally {System.out.println(future2 done...);}return efg;
});
调用 join() 后可以让程序等future1 和 future2 都运行完了之后再继续执行
CompletableFutureVoid completableFuture CompletableFuture.allOf(future1, future2);
completableFuture.join();
assertTrue(completableFuture.isDone());
System.out.println(all futures done...);
输出如下
future1 done...
future2 done...
all futures done...
anyOf() 方法不会等待所有的 CompletableFuture 都运行完成之后再返回只要有一个执行完成即可
CompletableFutureObject f CompletableFuture.anyOf(future1, future2);
System.out.println(f.get());
输出
可能是
future2 done...
efg
也可能是
future1 done...
abc
