CompletableFuture踩坑

简单记录一个基本的问题

private List fetchResult(List request){

...........

}

List ids=new ArrayList();//假设有10w个

List> patitionList=Lists.partition(ids,100);

List result=new ArrayList<>();

if(patitionList.size()==1){

//不用启动多线程，单线程执行其中的方法

result= fetchResult(patitionList.get(0));

}else{

List>> futureList=

patitionList.stream().map(id->{

CompletableFuture.supplyAsync(()->{

return fetchResult(id);

});

//线程池

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

CompletableFuture allFutures=CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0]));

CompletableFuture>> futureRes=allFutures.thenApply(v->{

allFutures.stream().map(CompletableFuture::join).collect(Collectors.toList());

});

allFutures.join();

List> resList=futureRes.join();

List flatMapResult=resList.stream().filter(CollectionUtils::isNotEmpty).flatMap(Collection::stream)..collect(Collectors.toList());

//接下来就在根据自己需要处理多线程的值

}

--------------------------------------------------------------

还有另外一种写法 需要注意 thenAccept。用不好容易出现线程安全的问题

List result=new ArrayList<>();--------一定要使用线程安全的集合 否则在thenAccept中发生线程不安全的问题

如果不想使用这种 可以使用上面那种，都在主线程中执行

CompletableFuture [] futures=

patitionList.stream().map(id->{

CompletableFuture.supplyAsync(()->{

return fetchResult(id);

});

//线程池

}).thenAccept(

list->{

result.addAll(list);

}

).toArray(CompletableFuture[]::new);

CompletableFuture.allOf(futures).join();