本文旨在解决使用CompletableFuture并行处理大型列表时遇到的性能瓶颈问题。通过移除导致串行执行的join操作，并提供两种等待所有任务完成的方法，帮助开发者充分利用多线程优势，显著提升数据处理速度。重点讲解如何正确地提交任务到线程池，并确保所有任务并行执行并最终完成，从而优化应用程序的性能。
在使用CompletableFuture处理大型列表时，并行执行可以显著提高处理速度。然而，不当的使用方式可能会导致看似并行，实则串行执行，从而达不到预期的性能提升效果。本文将针对这一问题，提供一种解决方案，并解释其背后的原理。
问题分析
原始代码中，在流式处理过程中使用了.map(CompletableFuture::join)。这个操作会导致主线程等待每个CompletableFuture完成，然后才能继续处理下一个CompletableFuture。因此，虽然每个CompletableFuture都在不同的线程中运行，但它们实际上是按顺序启动和完成的，导致并行处理失效。
解决方案
核心思想是先将所有CompletableFuture提交到线程池，然后等待所有任务完成，最后再收集结果。
代码示例
以下是修改后的代码示例：
灵云AI开放平台
灵云AI开放平台
150 查看详情
import com.google.common.collect.Lists;import java.util.List;import java.util.concurrent.CompletableFuture;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.stream.Collectors;public class CompletableFutureParallel { public static void main(String[] args) { int noOfCores = Runtime.getRuntime().availableProcessors(); ExecutorService service = Executors.newFixedThreadPool(noOfCores - 1); List list = List.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10); // 示例数据 // 将任务提交到线程池 List<CompletableFuture<List>> completableFutureList = Lists.partition(list, 2).stream() .map(item -> CompletableFuture.supplyAsync(() -> executeListPart(item), service)) .collect(Collectors.toList()); // 等待所有任务完成 completableFutureList.forEach(CompletableFuture::join); // 收集结果 List result = completableFutureList.stream() .map(CompletableFuture::join) // 再次join，确保获取到结果 .flatMap(List::stream) .collect(Collectors.toList()); System.out.println("Result: " + result); service.shutdown(); } // 模拟耗时操作 private static List executeListPart(List item) { try { Thread.sleep(1000); // 模拟处理时间 } catch (InterruptedException e) { Thread.currentThread().interrupt(); } System.out.println("Processing: " + item + " in thread: " + Thread.currentThread().getName()); return item.stream().map(i -> i * 2).collect(Collectors.toList()); // 示例操作 }}
代码解释
创建线程池: 使用Executors.newFixedThreadPool(noOfCores – 1)创建一个固定大小的线程池，线程数量通常设置为CPU核心数减1，以避免过度竞争。分割列表: 使用Lists.partition(list, 500)将大型列表分割成多个小列表，每个小列表作为一个任务提交到线程池。提交任务: 使用CompletableFuture.supplyAsync(() -> executeListPart(item), service)将每个小列表的处理任务提交到线程池。supplyAsync方法异步执行executeListPart方法，并返回一个CompletableFuture对象。收集CompletableFuture: 将所有CompletableFuture对象收集到一个列表中。等待任务完成: 使用completableFutureList.forEach(CompletableFuture::join)等待所有CompletableFuture对象完成。join方法会阻塞当前线程，直到对应的CompletableFuture完成。收集结果: 使用流式操作收集每个CompletableFuture的结果。
注意事项
异常处理: 在executeListPart方法中，需要妥善处理可能抛出的异常。可以将异常记录到日志中，或者向上抛出，由主线程处理。线程池管理: 在程序结束时，需要关闭线程池，释放资源。可以使用service.shutdown()方法关闭线程池。数据同步: 如果executeListPart方法需要访问共享数据，需要确保数据同步，避免出现线程安全问题。资源限制: 需要根据实际情况调整线程池大小和列表分割大小，避免资源耗尽。
另一种等待任务完成的方法
除了使用forEach(CompletableFuture::join)，还可以使用CompletableFuture.allOf等待所有任务完成。
CompletableFuture allFutures = CompletableFuture.allOf(completableFutureList.toArray(new CompletableFuture[0]));allFutures.join();
CompletableFuture.allOf方法返回一个新的CompletableFuture，当所有输入的CompletableFuture都完成时，该CompletableFuture才完成。
总结
通过移除导致串行执行的join操作，并使用forEach(CompletableFuture::join)或CompletableFuture.allOf等待所有任务完成，可以实现CompletableFuture的并行执行，从而显著提高大型列表的处理速度。 在实际应用中，需要根据具体情况调整线程池大小和列表分割大小，并注意异常处理和数据同步，以确保程序的正确性和性能。
以上就是并行执行CompletableFuture处理大型列表：优化性能的实用指南的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/750907.html
微信扫一扫 
支付宝扫一扫 
