在过滤器函数中使用CompletableFuture

Question

我有一个用例,我想根据我在元素上执行的网络调用过滤出列表中的一些元素.为了实现这一点,我使用流,过滤器和Completable Future.目标是执行异步执行,以便操作变得高效.下面提到伪代码.

public List<Integer> afterFilteringList(List<Integer> initialList){
   List<Integer> afterFilteringList =initialList.stream().filter(element -> {
        boolean valid = true;
        try{
            valid = makeNetworkCallAndCheck().get();
        } catch (Exception e) {

        }
        return valid;
    }).collect(Collectors.toList());

    return afterFilteringList;
}
public CompletableFuture<Boolean> makeNetworkCallAndCheck(Integer value){
   return CompletableFuture.completedFuture(resultOfNetWorkCall(value);
 }

我在这里遇到的问题是,我是否以异步方式进行此操作？(因为我在过滤器中使用'get'函数会阻止执行并使其仅为顺序)或者是否有更好的方法使用Java 8中的Completable Future和Filters以异步方式执行此操作.

Answer 1

当你get立即打电话时,你确实破坏了异步执行的好处.解决方案是在加入之前首先收集所有异步作业.

public List<Integer> afterFilteringList(List<Integer> initialList){
    Map<Integer,CompletableFuture<Boolean>> jobs = initialList.stream()
        .collect(Collectors.toMap(Function.identity(), this::makeNetworkCallAndCheck));
    return initialList.stream()
        .filter(element -> jobs.get(element).join())
        .collect(Collectors.toList());
}
public CompletableFuture<Boolean> makeNetworkCallAndCheck(Integer value){
   return CompletableFuture.supplyAsync(() -> resultOfNetWorkCall(value));
}

当然,该方法makeNetworkCallAndCheck也必须启动真正的异步操作.同步调用方法并返回a completedFuture是不够的.我在这里提供了一个简单的示例性异步操作,但是对于I/O操作,您可能希望提供自己的Executor,根据您想要允许的同时连接数量进行定制.