系列目录:
- Spring WebFlux运用中的思考与对比
- CompletableFuture与Spring的Sleuth结合工具类
- CommpetableFuture使用anyOf过程中的一些优化思考
- 结合CompletableFuture与Spring的Sleuth结合工具类与allOf以及anyOf
CompetableFuture的
上一篇我们讲述了如何将CompletableFuture与Spring Sleuth结合起来。这篇我们继续优化CompletableFuture。
CompletableFuture的allOf
首先我们看看allOf的定义:
1public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) { 2 // ... 3}
这个方法接受若干个返回不同类型的CompletableFuture为参数, 返回一个返回为空(Void)的CompletableFuture。也就是说,这个方法其实就是返回一个在所有参数完成之后也完成的返回为空(Void)的CompletableFuture,也就是充当一个signaling device
这个方法很好,尤其是并发获取多种io的结果的时候。但是用这个方法,带来了很多不便,最大的不便就是,返回是Void,而不是所有的参数的返回。这样导致我们,需要在聚合这些结果的那个服务方法里面,把最终结果封装好,否则,获取的就是一个Void。举个例子:
假设我的一个服务方法的返回是多个接口在使用,这个方法需要同时调用三个io等待他们都返回时,利用这三个io的返回,拼装成接口需要的字段。对于这个场景,我们可以有两种写法,第一种是基于回调的写法,第二种是基于返回的写法,两种都OK,看个人习惯,我个人倾向于基于返回的写法,这样代码是瀑布式的,基于回调的会导致多层嵌套,导致代码可读性降低。
** 结果类:**
1@Data 2@Builder 3@NoArgsConstructor 4@AllArgsConstructor 5public static class Result { 6 private String string; 7 private List<String> strings; 8 private List<Integer> integers; 9}
** 基于回调:**
1public static void baseOnCallBack(CompletableFuture<Result> resultCompletableFuture) { 2 CompletableFuture<List<String>> result1 = CompletableFuture.supplyAsync(() -> { 3 //模拟io 4 try { 5 TimeUnit.MILLISECONDS.sleep(ThreadLocalRandom.current().nextLong(1000)); 6 } catch (InterruptedException e) { 7 e.printStackTrace(); 8 } 9 return Lists.newArrayList("a", "b", "c"); 10 }); 11 CompletableFuture<List<Integer>> result2 = CompletableFuture.supplyAsync(() -> { 12 //模拟io 13 try { 14 TimeUnit.MILLISECONDS.sleep(ThreadLocalRandom.current().nextLong(1000)); 15 } catch (InterruptedException e) { 16 e.printStackTrace(); 17 } 18 return Lists.newArrayList(1, 2, 3); 19 }); 20 CompletableFuture<String> result3 = CompletableFuture.supplyAsync(() -> { 21 //模拟io 22 try { 23 TimeUnit.MILLISECONDS.sleep(ThreadLocalRandom.current().nextLong(1000)); 24 } catch (InterruptedException e) { 25 e.printStackTrace(); 26 } 27 return "hash-test"; 28 }); 29 30 CompletableFuture.allOf(result1, result2, result3).thenAcceptAsync(v -> { 31 resultCompletableFuture.complete(Result.builder() 32 //一定存在的,因为已经完成了 33 .string(result3.join()) 34 .strings(result1.join()) 35 .integers(result2.join()) 36 .build()); 37 }); 38}
** 基于返回:**
1public static CompletableFuture<Result> baseOnReturn() { 2 CompletableFuture completableFuture = new CompletableFuture(); 3 CompletableFuture<List<String>> result1 = CompletableFuture.supplyAsync(() -> { 4 //模拟io 5 try { 6 TimeUnit.MILLISECONDS.sleep(ThreadLocalRandom.current().nextLong(1000)); 7 } catch (InterruptedException e) { 8 e.printStackTrace(); 9 } 10 return Lists.newArrayList("a", "b", "c"); 11 }); 12 CompletableFuture<List<Integer>> result2 = CompletableFuture.supplyAsync(() -> { 13 //模拟io 14 try { 15 TimeUnit.MILLISECONDS.sleep(ThreadLocalRandom.current().nextLong(1000)); 16 } catch (InterruptedException e) { 17 e.printStackTrace(); 18 } 19 return Lists.newArrayList(1, 2, 3); 20 }); 21 CompletableFuture<String> result3 = CompletableFuture.supplyAsync(() -> { 22 //模拟io 23 try { 24 TimeUnit.MILLISECONDS.sleep(ThreadLocalRandom.current().nextLong(1000)); 25 } catch (InterruptedException e) { 26 e.printStackTrace(); 27 } 28 return "hash-test"; 29 }); 30 CompletableFuture.allOf(result1, result2, result3).thenAcceptAsync(v -> { 31 completableFuture.complete(Result.builder() 32 //一定存在的,因为已经完成了 33 .string(result3.join()) 34 .strings(result1.join()) 35 .integers(result2.join()) 36 .build()); 37 }); 38 return completableFuture; 39}
基于回调的接口使用结果:
1CompletableFuture completableFuture = new CompletableFuture(); 2baseOnCallBack(completableFuture); 3completableFuture = completableFuture.thenAcceptAsync(result -> { 4 System.out.println("baseOnCallback: " + result); 5});
基于返回的接口使用结果:
1CompletableFuture<Void> voidCompletableFuture = baseOnReturn().thenAcceptAsync(result -> { 2 System.out.println("baseOnReturn: " + result); 3});
可以看出,一层嵌套也是基于返回的代码看上去更优雅。
我们再来思考下,如果allOf中的所有CompletableFuture都返回的是同一个类型的结果,例如String,那么可不可以让allOf直接返回List<String>呢?
我们可以将一个allOf变成多个allOf这么实现:
1public static <T> CompletableFuture<List<T>> allOf(Collection<CompletableFuture<T>> futures) { 2 return futures.stream().collect(Collectors.collectingAndThen( 3 Collectors.toList(), 4 l -> CompletableFuture.allOf(l.toArray(new CompletableFuture[0])) 5 .thenApply(v -> l.stream().map(CompletableFuture::join).collect(Collectors.toList())) 6 ) 7 ); 8}