搜索[AIPC技术]结果-低调大师优秀个人博客

☕【Java技术指南】教你如何使用异步神器CompletableFuture

前提概要在java8以前，我们使用java的多线程编程,一般是通过Runnable中的run方法来完成,这种方式，有个很明显的缺点,就是，没有返回值。这时候，大家可能会去尝试使用Callable中的call方法，然后用Future返回结果，如下: public static void main(String[] args) throws Exception { ExecutorService executor = Executors.newSingleThreadExecutor(); Future<String> stringFuture = executor.submit(new Callable<String>() { @Override public String call() throws Exception { Thread.sleep(2000); return "async thread"; } }); Thread.sleep(1000); System.out.println("main thread"); System.out.println(stringFuture.get()); } 通过观察控制台,我们发现先打印 main thread ,一秒后打印 async thread，似乎能满足我们的需求。但仔细想我们发现一个问题，当调用future的get()方法时，当前主线程是堵塞的，这好像并不是我们想看到的。另一种获取返回结果的方式是先轮询,可以调用isDone，等完成再获取，但这也不能让我们满意. 很多个异步线程执行时间可能不一致,我的主线程业务不能一直等着,这时候我可能会想要只等最快的线程执行完或者最重要的那个任务执行完,亦或者我只等1秒钟,至于没返回结果的线程我就用默认值代替. 我两个异步任务之间执行独立,但是第二个依赖第一个的执行结果. java8的CompletableFuture,就在这混乱且不完美的多线程江湖中闪亮登场了.CompletableFuture让Future的功能和使用场景得到极大的完善和扩展,提供了函数式编程能力,使代码更加美观优雅,而且可以通过回调的方式计算处理结果,对异常处理也有了更好的处理手段. CompletableFuture源码中有四个静态方法用来执行异步任务: 创建任务 public static CompletableFuture supplyAsync(Supplier supplier){..} public static CompletableFuture supplyAsync(Supplier supplier,Executor executor){..} public static CompletableFuture<Void> runAsync(Runnable runnable){..} public static CompletableFuture<Void> runAsync(Runnable runnable,Executor executor){..} 如果有多线程的基础知识，我们很容易看出，run开头的两个方法，用于执行没有返回值的任务，因为它的入参是Runnable对象。而supply开头的方法显然是执行有返回值的任务了，至于方法的入参，如果没有传入Executor对象将会使用ForkJoinPool.commonPool() 作为它的线程池执行异步代码.在实际使用中,一般我们使用自己创建的线程池对象来作为参数传入使用，这样速度会快些. 执行异步任务的方式也很简单,只需要使用上述方法就可以了: CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { //....执行任务 return "hello"; }, executor) 接下来看一下获取执行结果的几个方法。 V get(); V get(long timeout,Timeout unit); T getNow(T defaultValue); T join(); 上面两个方法是Future中的实现方式，get()会堵塞当前的线程，这就造成了一个问题,如果执行线程迟迟没有返回数据,get()会一直等待下去,因此,第二个get()方法可以设置等待的时间. getNow()方法比较有意思，表示当有了返回结果时会返回结果，如果异步线程抛了异常会返回自己设置的默认值. 接下来以一些场景的实例来介绍一下CompletableFuture中其他一些常用的方法 thenAccept() public CompletionStage<Void> thenAccept(Consumer<? super T> action); public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action); public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action,Executor executor); 功能：当前任务正常完成以后执行，当前任务的执行结果可以作为下一任务的输入参数，无返回值. 场景：执行任务A,同时异步执行任务B，待任务B正常返回后，B的返回值执行任务C，任务C无返回值 CompletableFuture<String> futureA = CompletableFuture.supplyAsync(() -> "任务A"); CompletableFuture<String> futureB = CompletableFuture.supplyAsync(() -> "任务B"); CompletableFuture<String> futureC = futureB.thenApply(b -> { System.out.println("执行任务C."); System.out.println("参数:" + b);//参数:任务B return "a"; }); thenRun(..) public CompletionStage<Void> thenRun(Runnable action); public CompletionStage<Void> thenRunAsync(Runnable action); public CompletionStage<Void> thenRunAsync(Runnable action,Executor executor); 功能：对不关心上一步的计算结果，执行下一个操作场景：执行任务A,任务A执行完以后,执行任务B,任务B不接受任务A的返回值(不管A有没有返回值),也无返回值 CompletableFuture<String> futureA = CompletableFuture.supplyAsync(() -> "任务A"); futureA.thenRun(() -> System.out.println("执行任务B")); thenApply(..) public CompletableFuture thenApply(Function<? super T,? extends U> fn) public CompletableFuture thenApplyAsync(Function<? super T,? extends U> fn) public CompletableFuture thenApplyAsync(Function<? super T,? extends U> fn, Executor executor) 功能：当前任务正常完成以后执行，当前任务的执行的结果会作为下一任务的输入参数,有返回值场景：多个任务串联执行,下一个任务的执行依赖上一个任务的结果,每个任务都有输入和输出异步执行任务A，当任务A完成时使用A的返回结果resultA作为入参进行任务B的处理，可实现任意多个任务的串联执行 CompletableFuture<String> futureA = CompletableFuture.supplyAsync(() -> "hello"); CompletableFuture<String> futureB = futureA.thenApply(s->s + " world"); CompletableFuture<String> future3 = futureB.thenApply(String::toUpperCase); System.out.println(future3.join()); 上面的代码，我们当然可以先调用future.join()先得到任务A的返回值,然后再拿返回值做入参去执行任务B,而thenApply的存在就在于帮我简化了这一步,我们不必因为等待一个计算完成而一直阻塞着调用线程，而是告诉CompletableFuture你啥时候执行完就啥时候进行下一步. 就把多个任务串联起来了. thenCombine(..) thenAcceptBoth(..) runAfterBoth(..) 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) 功能：结合两个CompletionStage的结果，进行转化后返回场景：需要根据商品id查询商品的当前价格,分两步,查询商品的原始价格和折扣,这两个查询相互独立,当都查出来的时候用原始价格乘折扣,算出当前价格. 使用方法:thenCombine(..) CompletableFuture<Double> futurePrice = CompletableFuture.supplyAsync(() -> 100d); CompletableFuture<Double> futureDiscount = CompletableFuture.supplyAsync(() -> 0.8); CompletableFuture<Double> futureResult = futurePrice.thenCombine(futureDiscount, (price, discount) -> price * discount); System.out.println("最终价格为:" + futureResult.join()); //最终价格为:80.0 thenCombine(..)是结合两个任务的返回值进行转化后再返回,那如果不需要返回呢,那就需要 thenAcceptBoth(..),同理,如果连两个任务的返回值也不关心呢,那就需要runAfterBoth了,如果理解了上面三个方法,thenApply,thenAccept,thenRun,这里就不需要单独再提这两个方法了,只在这里提一下. thenCompose(..) public CompletableFuture thenCompose(Function<? super T,? extends CompletionStage> fn) public CompletableFuture thenComposeAsync(Function<? super T,? extends CompletionStage> fn) public CompletableFuture thenComposeAsync(Function<? super T,? extends CompletionStage> fn, Executor executor) 功能：这个方法接收的输入是当前的CompletableFuture的计算值，返回结果将是一个新的CompletableFuture 这个方法和thenApply非常像,都是接受上一个任务的结果作为入参,执行自己的操作,然后返回.那具体有什么区别呢? thenApply():它的功能相当于将CompletableFuture<T>转换成CompletableFuture,改变的是同一个CompletableFuture中的泛型类型 thenCompose():用来连接两个CompletableFuture，返回值是一个新的CompletableFuture CompletableFuture<String> futureA = CompletableFuture.supplyAsync(() -> "hello"); CompletableFuture<String> futureB = futureA.thenCompose(s -> CompletableFuture.supplyAsync(() -> s + "world")); CompletableFuture<String> future3 = futureB.thenCompose(s -> CompletableFuture.supplyAsync(s::toUpperCase)); System.out.println(future3.join()); applyToEither(..) acceptEither(..) runAfterEither(..) public CompletionStage applyToEither(CompletionStage<? extends T> other,Function<? super T, U> fn); public CompletionStage applyToEitherAsync(CompletionStage<? extends T> other,Function<? super T, U> fn); public CompletionStage applyToEitherAsync(CompletionStage<? extends T> other,Function<? super T, U> fn,Executor executor); 功能:执行两个CompletionStage的结果,那个先执行完了,就是用哪个的返回值进行下一步操作场景:假设查询商品a,有两种方式,A和B,但是A和B的执行速度不一样,我们希望哪个先返回就用那个的返回值. 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); System.out.println(futureC.join()); //结果:通过方式A获取商品a 同样的道理,applyToEither的兄弟方法还有acceptEither(),runAfterEither(),我想不需要我解释你也知道该怎么用了. exceptionally(..) public CompletionStage<T> exceptionally(Function<Throwable, ? extends T> fn); 功能:当运行出现异常时,调用该方法可进行一些补偿操作,如设置默认值. 场景:异步执行任务A获取结果,如果任务A执行过程中抛出异常,则使用默认值100返回. CompletableFuture<String> futureA = CompletableFuture. supplyAsync(() -> "执行结果:" + (100 / 0)) .thenApply(s -> "futureA result:" + s) .exceptionally(e -> { System.out.println(e.getMessage()); //java.lang.ArithmeticException: / by zero return "futureA result: 100"; }); CompletableFuture<String> futureB = CompletableFuture. supplyAsync(() -> "执行结果:" + 50) .thenApply(s -> "futureB result:" + s) .exceptionally(e -> "futureB result: 100"); System.out.println(futureA.join());//futureA result: 100 System.out.println(futureB.join());//futureB result:执行结果:50 上面代码展示了正常流程和出现异常的情况,可以理解成catch,根据返回值可以体会下. whenComplete(..) public CompletionStage<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); 功能:当CompletableFuture的计算结果完成，或者抛出异常的时候，都可以进入whenComplete方法执行,举个栗子 CompletableFuture<String> futureA = CompletableFuture. supplyAsync(() -> "执行结果:" + (100 / 0)) .thenApply(s -> "apply result:" + s) .whenComplete((s, e) -> { if (s != null) { System.out.println(s);//未执行 } if (e == null) { System.out.println(s);//未执行 } else { System.out.println(e.getMessage());//java.lang.ArithmeticException: / by zero } }) .exceptionally(e -> { System.out.println("ex"+e.getMessage()); //ex:java.lang.ArithmeticException: / by zero return "futureA result: 100"; }); System.out.println(futureA.join());//futureA result: 100 根据控制台,我们可以看出执行流程是这样,supplyAsync->whenComplete->exceptionally,可以看出并没有进入thenApply执行,原因也显而易见,在supplyAsync中出现了异常,thenApply只有当正常返回时才会去执行.而whenComplete不管是否正常执行,还要注意一点,whenComplete是没有返回值的. 上面代码我们使用了函数式的编程风格并且先调用whenComplete再调用exceptionally,如果我们先调用exceptionally,再调用whenComplete会发生什么呢,我们看一下: 复制代码 CompletableFuture<String> futureA = CompletableFuture. supplyAsync(() -> "执行结果:" + (100 / 0)) .thenApply(s -> "apply result:" + s) .exceptionally(e -> { System.out.println("ex:"+e.getMessage()); //ex:java.lang.ArithmeticException: / by zero return "futureA result: 100"; }) .whenComplete((s, e) -> { if (e == null) { System.out.println(s);//futureA result: 100 } else { System.out.println(e.getMessage());//未执行 } }) ; System.out.println(futureA.join());//futureA result: 100 代码先执行了exceptionally后执行whenComplete,可以发现,由于在exceptionally中对异常进行了处理,并返回了默认值,whenComplete中接收到的结果是一个正常的结果,被exceptionally美化过的结果,这一点需要留意一下. handle(..) public CompletionStage handle(BiFunction<? super T, Throwable, ? extends U> fn); public CompletionStage handleAsync(BiFunction<? super T, Throwable, ? extends U> fn); public CompletionStage handleAsync(BiFunction<? super T, Throwable, ? extends U> fn,Executor executor); 功能:当CompletableFuture的计算结果完成，或者抛出异常的时候，可以通过handle方法对结果进行处理 CompletableFuture<String> futureA = CompletableFuture. supplyAsync(() -> "执行结果:" + (100 / 0)) .thenApply(s -> "apply result:" + s) .exceptionally(e -> { System.out.println("ex:" + e.getMessage()); //java.lang.ArithmeticException: / by zero return "futureA result: 100"; }) .handle((s, e) -> { if (e == null) { System.out.println(s);//futureA result: 100 } else { System.out.println(e.getMessage());//未执行 } return "handle result:" + (s == null ? "500" : s); }); System.out.println(futureA.join());//handle result:futureA result: 100 通过控制台,我们可以看出,最后打印的是handle result:futureA result: 100,执行exceptionally后对异常进行了"美化",返回了默认值,那么handle得到的就是一个正常的返回,我们再试下,先调用handle再调用exceptionally的情况. CompletableFuture<String> futureA = CompletableFuture. supplyAsync(() -> "执行结果:" + (100 / 0)) .thenApply(s -> "apply result:" + s) .handle((s, e) -> { if (e == null) { System.out.println(s);//未执行 } else { System.out.println(e.getMessage());//java.lang.ArithmeticException: / by zero } return "handle result:" + (s == null ? "500" : s); }) .exceptionally(e -> { System.out.println("ex:" + e.getMessage()); //未执行 return "futureA result: 100"; }); System.out.println(futureA.join());//handle result:500 根据控制台输出,可以看到先执行handle,打印了异常信息,并对接过设置了默认值500,exceptionally并没有执行,因为它得到的是handle返回给它的值,由此我们大概推测handle和whenComplete的区别都是对结果进行处理,handle有返回值,whenComplete没有返回值由于1的存在,使得handle多了一个特性,可在handle里实现exceptionally的功能 allOf(..) anyOf(..) public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) allOf:当所有的CompletableFuture都执行完后执行计算 anyOf:最快的那个CompletableFuture执行完之后执行计算场景二:查询一个商品详情,需要分别去查商品信息,卖家信息,库存信息,订单信息等,这些查询相互独立,在不同的服务上,假设每个查询都需要一到两秒钟,要求总体查询时间小于2秒. public static void main(String[] args) throws Exception { ExecutorService executorService = Executors.newFixedThreadPool(4); long start = System.currentTimeMillis(); CompletableFuture<String> futureA = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(1000 + RandomUtils.nextInt(1000)); } catch (InterruptedException e) { e.printStackTrace(); } return "商品详情"; },executorService); CompletableFuture<String> futureB = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(1000 + RandomUtils.nextInt(1000)); } catch (InterruptedException e) { e.printStackTrace(); } return "卖家信息"; },executorService); CompletableFuture<String> futureC = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(1000 + RandomUtils.nextInt(1000)); } catch (InterruptedException e) { e.printStackTrace(); } return "库存信息"; },executorService); CompletableFuture<String> futureD = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(1000 + RandomUtils.nextInt(1000)); } catch (InterruptedException e) { e.printStackTrace(); } return "订单信息"; },executorService); CompletableFuture<Void> allFuture = CompletableFuture.allOf(futureA, futureB, futureC, futureD); allFuture.join(); System.out.println(futureA.join() + futureB.join() + futureC.join() + futureD.join()); System.out.println("总耗时:" + (System.currentTimeMillis() - start)); }

2021-07-31

OSChina还是不是技术人的交流社区？

当初的OSChina和CSDN是国内两大技术社区，为什么没有选择CSDN是因为它的商业化过于严重，首页基本是广告或者其他商业实体的技术推广，这不是技术人的社区，是在获得了技术人的流量之后，出卖了作为技术社区的本质

2021-07-21

oCPX简介——广告界的“无人驾驶”技术

需要对接很多中小广告主，他们技术水平不高，回传转化数据流程容易出问题。

2021-07-15

边缘计算：最快数据处理背后的技术

边缘计算简介边缘计算是分布式信息技术架构的一种实践，其中客户端的数据在网络边缘处理，尽可能接近原始来源。

2021-07-12

容器技术，带着云原生百米冲刺

适应云计算的发展，云原生技术已经成为加快企业数字化转型的一个不折不扣的风向标。而以容器为代表的云原生技术正在成为释放云价值的最短路径。

2021-06-23

Tapdata 数据库实时同步的技术要点

Tapdata 专注于实时数据的处理技术，在数据库迁移和同步方面，Tapdata 的表现非常优秀，实时、多元、异构，尤其在关系数据库到非关系数据库之间的双向同步方面，无论是从操作上，还是效率上，都体现了业界领先的水平

2021-06-04

前端技术之实用的 JavaScript 调试小技巧

除了console.log，debugger就是另一个我很喜欢的快速调试的工具，将debugger加入代码之后，Chrome会自动在插入它的地方停止，很像C或者Java里面打断点。你也可以在一些条件控制中插入该调试语句，譬如： [JavaScript]纯文本查看复制代码 ? 1 2 3 if (thisThing) { debugger; } 将Objects以表格形式展示有时候我们需要看一些复杂的对象的详细信息，最简单的方法就是用console.log然后展示成一个列表状，上下滚动进行浏览。不过似乎用console.table展示成列表会更好呦，大概是介个样子： [JavaScript]纯文本查看复制代码 ? 1 2 3 4 5 6 7 var animals = [ { animal: 'Horse', name: 'Henry', age: 43 }, { animal: 'Dog', name: 'Fred', age: 13 }, { animal: 'Cat', name: 'Frodo', age: 18 } ]; console.table(animals); 多屏幕尺寸测试 Chrome有一个非常诱人的功能就是能够模拟不同设备的尺寸，在Chrome的Inspector中点击toggle device mode按钮，然后就可以在不同的设备屏幕尺寸下进行调试咯：在Console快速选定DOM元素在Elements选择面板中选择某个DOM元素然后在Console中使用该元素也是非常常见的一个操作，Chrome Inspector会缓存最后5个DOM元素在它的历史记录中，你可以用类似于Shell中的$0等方式来快速关联到元素。譬如下图的列表中有‘item-4′, ‘item-3’, ‘item-2’, ‘item-1’, ‘item-0’这几个元素，你可以这么使用：获取某个函数的调用追踪记录 JavaScript框架极大方便了我们的开发，但是也会带来大量的预定义的函数，譬如创建View的、绑定事件的等等，这样我们就不容易追踪我们自定义函数的调用过程了。虽然JavaScript不是一个非常严谨的语言，有时候很难搞清楚到底发生了啥，特别是当你需要审阅其他人的代码的时候。这时候console.trace就要起作用咯，它可以帮你进行函数调用的追踪。譬如下面的代码中我们要追踪出car对象中对于funcZ的调用过程： [JavaScript]纯文本查看复制代码 ? 01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 var car; var func1 = function() { func2(); } var func2 = function() { func4(); } var func3 = function() { } var func4 = function() { car = new Car(); car.funcX(); } var Car = function() { this.brand = ‘volvo’; this.color = ‘red’; this.funcX = function() { this.funcY(); } this.funcY = function() { this.funcZ(); } this.funcZ = function() { console.trace(‘trace car’) } } func1(); 这边就可以清晰地看出func1调用了func2，然后调用了func4，func4创建了Car的实例然后调用了car.funcX。格式化被压缩的代码有时候在生产环境下我们发现了一些莫名奇妙的问题，然后忘了把sourcemaps放到这台服务器上，或者在看别人家的网站的源代码的时候，结果就看到了一坨不知道讲什么的代码，就像下图。Chrome为我们提供了一个很人性化的反压缩工具来增强代码的可读性，大概这么用：快速定位调试函数当我们想在函数里加个断点的时候，一般会选择这么做：在Inspector中找到指定行，然后添加一个断点在脚本中添加一个debugger调用不过这两种方法都存在一个小问题就是都要到对应的脚本文件中然后再找到对应的行，这样会比较麻烦。这边介绍一个相对快捷点的方法，就是在console中使用debug(funcName)然后脚本会在指定到对应函数的地方自动停止。这种方法有个缺陷就是无法在私有函数或者匿名函数处停止，所以还是要因时而异： [JavaScript]纯文本查看复制代码 ? 01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 var func1 = function() { func2(); }; var Car = function() { this.funcX = function() { this.funcY(); } this.funcY = function() { this.funcZ(); } } var car = new Car(); 禁止不相关的脚本运行当我们开发现代网页的时候都会用一些第三方的框架或者库，它们几乎都是经过测试并且相对而言Bug较少的。不过当我们调试我们自己的脚本的时候也会一不小心跳到这些文件中，引发额外的调试任务。解决方案呢就是禁止这部分不需要调试的脚本运行。在较复杂的调试情况下发现关键元素在一些复杂的调试环境下我们可能要输出很多行的内容，这时候我们习惯性的会用console.log, console.debug, console.warn, console.info, console.error这些来进行区分，然后就可以在Inspector中进行过滤。不过有时候我们还是希望能够自定义显示样式，你可以用CSS来定义个性化的信息样式： [JavaScript]纯文本查看复制代码 ? 01 02 03 04 05 06 07 08 09 10 console.todo = function(msg) { console.log(‘ % c % s % s % s‘, ‘color: yellow; background - color: black;’, ‘–‘, msg, ‘–‘); } console.important = function(msg) { console.log(‘ % c % s % s % s’, ‘color: brown; font - weight: bold; text - decoration: underline;’, ‘–‘, msg, ‘–‘); } console.todo(“This is something that’ s need to be fixed”); console.important(‘This is an important message’); 在console.log()中你可以使用%s来代表一个字符串 ,?%i?来代表数字，以及?%c?来代表自定义的样式。监测指定函数的调用与参数在Chrome中可以监测指定函数的调用情况以及参数： [JavaScript]纯文本查看复制代码 ? 1 2 var func1 = function(x, y, z) { }; 这种方式能够让你实时监控到底啥参数被传入到了指定函数中。 Console中使用$进行元素查询在Console中也可以使用来进行类似于querySelector那样基于CSS选择器的查询，(‘css-selector’) 会返回满足匹配的第一个元素，而$$(‘css-selector’) 会返回全部匹配元素。注意，如果你会多次使用到元素，那么别忘了将它们存入变量中。 Postman 很多人习惯用Postman进行API调试或者发起Ajax请求，不过别忘了你浏览器自带的也能做这个，并且你也不需要担心啥认证啊这些，因为Cookie都是自带帮你传送的，这些只要在network这个tab里就能进行，大概这样子： DOM变化检测 DOM有时候还是很操蛋的，有时候压根不知道啥时候就变了，不过Chrome提供了一个小的功能就是当DOM发生变化的时候它会提醒你，你可以监测属性变化等等：

2021-06-03