> ## Content Index > Fetch the complete content index at: https://yonglun.me/llms.txt > Use this file to discover other available public pages before exploring further. # Java 并行(Parallelism)与并发(Concurrency)处理 - URL: https://yonglun.me/java-parallel/ - Published: 2023-11-24T10:38:33.000Z - Updated: 2025-04-22T07:01:50.000Z - Description: 使用Java8的并行处理能力，可以提升我们程序的执行效率，充分利用计算机的硬件资源。 Parallel Stream 利用CPU的多内核来进行并行处理，并行度由CPU的内核数量决定。 CompletableFuture 使用线程池来进行并发处理，并发数由线程池的线程数量决定。 - Author: Yonglun - Tags: Java, Parallel, Concurrency, Parallelism, Sequential, #lang-zh # 1\. 前言 最近项目团队遇到一个问题，一个用Java开发的定时任务执行效率达不到要求，凌晨2点开始运行，执行到早上9点客户上班了，程序还没处理完，影响了客户白天正常的业务。和负责开发的工程师聊了下，他反馈主要原因有三个： 1. 需要处理的数据量较大，大概有6000条。 2. 每条数据处理时间需要5秒。 3. 处理是串行的，处理完一条才开始处理第二条。 和他一起分析了下，可以从2、3入手手去提升。由于程序执行时服务器的资源情况是比较空闲的，所以将串行处理改为并行执应该能解决这个问题。 从Java8 开始，Java已经提供了很多针对并行处理的高级能力，能让我们轻松地编写并行处理的代码，充分利用现代计算机的硬件资源。 下面我们将以一个订单列表的处理为例子，说明下如何利用Java8的并行处理能力提升程序的执行效率。 # 2\. 并行(Parallelism) 与并发(Concurrency)处理 ## 2.1 串行执行 假设我们由A、B、C、D 4长订单要处理，每张订单的处理时间为1秒，串行执行情况下，处理完成一张订单后再开始另一张订单的处理，所以一共需要4秒完成所有的处理。 ![](https://yonglun.me/content/images/2023/11/---3.png) Sequential 如果订单处理过程中有很多I/O 操作，那么处理中需要有很多I/O 等待时间，那么这些时间都浪费掉了。 另外如果执行的机器上有多个CPU/内核，那么其他的CPU资源也浪费掉了。 ## 2.2 并行执行 并行执行是将处理分配到每个CPU上，每个CPU 同一时间处理一张订单。 从下图可见，如果每张订单的处理时间为1秒，执行的机器上有2个CPU，忽略并行任务分配的开销，那么一共需要约2秒完成所有的处理。 ![](https://yonglun.me/content/images/2023/11/---1.png) Parallelism 所以并行执行可以充分利用机器上的多CPU内核资源，提高执行效率。 ## 2.3 并发执行 并发执行是利用多线程执行能力，并发处理多个任务。线程可以被操作系统分配到多个CPU内核上执行，线程被操作系统调度进行执行和等待，从而达到多个任务行发执行的效果。 从下图可见，如果每张订单的处理时间为1秒，使用4个线程来执行，忽略线程分配和调度的开销，那么一共需要约1秒完成所有的处理。 ![](https://yonglun.me/content/images/2023/11/---2.png) Concurrency # 3\. 并行处理样例代码 ## 3.1 订单处理服务 ### 订单实体 ```Java /** * 订单 */ public class Order { private String id; public Order(String id){ this.id = id; } public String getId(){ return id; } public void setId(String id){ this.id = id; } } ``` ### 订单处理服务 模拟每个订单处理需要1秒时间 ```Java /** * 订单服务 */ public class OrderService { /** * 订单处理 * @param order 订单 * @return 处理结果 */ public static ProcessResult Process(Order order) { ProcessResult result = new ProcessResult(); result.setOrderId(order.getId()); try { //模拟订单处理需要1秒时间 Thread.sleep(1000L); result.setCode(ResultCode.SUCCESS); result.setMessage("Process Success"); } catch (Exception ex) { //记录订单处理异常日志 result.setCode(ResultCode.FAILURE); result.setMessage("Process Failure"); } return result; } } /** * 处理结果 */ public class ProcessResult { private int code; private String message; private String orderId; public int getCode() { return code; } public void setCode(int code) { this.code = code; } public String getMessage() { return message; } public void setMessage(String message) { this.message = message; } public String getOrderId() { return orderId; } public void setOrderId(String orderId) { this.orderId = orderId; } } ``` ## 3.2 准备测试数据 我们准备了20条订单数据来进行测试 ```Java /** * 准备用于测试的订单列表 * */ private static List prepareOrders(){ List orders = new ArrayList(); for (int i = 0; i < 20; i++) { Order order = new Order(UUID.randomUUID().toString()); orders.add(order); } return orders; } ``` ## 3.3 串行处理 这个是一个对订单列表进行串行处理的样例代码。 ```Java /** * 串行执行 */ private static void runSerial(){ //开始时间 long start = System.currentTimeMillis(); //准备用于测试的订单 List orders = prepareOrders(); //处理 List results = orders.stream() .map(OrderService::Process) .collect(Collectors.toList()); //结束时间 long end = System.currentTimeMillis(); //输出处理时间 System.out.printf("串行处理订单，处理时间为 %s ms%n", end - start); // System.out.println("处理状态为: " + results); } ``` ## 3.4 Parallel Stream 并行处理 Parallel Stream 是Java8 的一个并行处理功能，可以利用多核CPU并行对Stream中的数据进行处理。并行度由执行程序机器的CPU内核数量决定。例如在一台有8个内核的机器上，Parallel Stream 的并行度是8。 ```Java /** * Parallel Stream 并行处理 */ private static void runParallel(){ //开始时间 long start = System.currentTimeMillis(); //准备用于测试的订单 List orders = prepareOrders(); //处理 List results = orders.stream() .parallel() .map(OrderService::Process) .collect(Collectors.toList()); //结束时间 long end = System.currentTimeMillis(); //输出处理时间 System.out.printf("Parallel Stream 并行处理订单，处理时间为 %s ms%n", end - start); //System.out.println("处理状态为: " + results); } ``` ## 3.5 CompletableFuture 多线程并发处理 CompletableFuture 是Java8中对多线程处理的一个高级封装，可以使用线程池来并行执行处理任务。 ```Java /** * CompletableFuture 多线程并发处理 * @param executor 线程池 */ private static void runCompletableFuture(Executor executor){ //开始时间 long start = System.currentTimeMillis(); //准备用于测试的订单 List orders = prepareOrders(); //处理 List> futureResults = orders.stream() .map(o -> CompletableFuture.supplyAsync( () -> OrderService.Process(o),executor) ).collect(Collectors.toList()); //收集结果 List results = futureResults.stream() .map(CompletableFuture::join) .collect(Collectors.toList()); //结束时间 long end = System.currentTimeMillis(); //输出处理时间 System.out.printf("CompletableFuture 多线程并发处理订单，处理时间为 %s ms%n", end - start); //System.out.println("处理状态为: " + results); } ``` ## 3.6 测试程序 我们将上面的代码集成起来，编写一个测试程序，分别按三种方式对订单进行处理，并输出各自的处理时间。 ```Java public static void main(String[] args) { //准备一个线程池，线程数量为20 Executor executor = Executors.newFixedThreadPool(20); //串行执行 runSerial(); //Parallel Stream 并行处理 runParallel(); //CompletableFuture 多线程并发处理 runCompletableFuture(executor); System.exit(0); } ``` ## 3.7 运行结果 在一台8个CPU内核的机器上运行该程序，输出结果如下： ``` 串行处理订单，处理时间为 20150 ms Parallel Stream 并行处理订单，处理时间为 3016 ms CompletableFuture 多线程并发处理订单，处理时间为 1013 ms Process finished with exit code 0 ``` 对20条订单，每条订单处理时间为1秒，运行结果如下： 串行处理需要20秒。 > 这是没有使用任何并行处理的情况，每条订单的处理结束后，再处理下一条，所以一共耗时20秒。 Parallel Stream 并行处理需要3秒。 > 因为机器有8个内核，所以对于Parallel Stream并行处理时并行度为8，20个订单需要分3次才能处理完成，所以一共耗时3秒。 CompletableFuture 多线程并发处理需要1秒 > 我们使用了一个有20个线程的线程池来执行，所以20个订单的处理一共耗时1秒。 # 4\. 总结 从上面的样例代码可见: - 使用Java8的并行处理能力，可以提升我们程序的执行效率，充分利用计算机的硬件资源。 - Parallel Stream 利用CPU的多内核来进行并行处理，并行度由CPU的内核数量决定。 - CompletableFuture 使用线程池来进行并发处理，并发数由线程池的线程数量决定。