Java使用多线程处理未知任务数方案

知道任务个数,你可以定义好线程数规则,生成线程数去跑

代码说明:

虚拟线程池:

  • 使用 Executors.newVirtualThreadPerTaskExecutor() 创建虚拟线程池,每个任务将分配一个虚拟线程来执行。

提交任务并返回结果:

  • 每个任务通过 CompletableFuture.supplyAsync() 提交,任务会返回一个结果(这里是字符串,模拟了任务的处理结果)。

  • 每个 CompletableFuture 都会保存任务的返回值。

等待所有任务完成:

  • 使用 CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])) 等待所有的 CompletableFuture 完成。allOf.join() 会阻塞当前线程,直到所有任务完成。

收集结果:

  • 使用 Java 8 的 stream() 方法和 Collectors.toList() 来收集所有任务的结果,并将它们合并到一个 List 中。
  • CompletableFuture::join 会获取每个任务的结果,并且如果任务有异常,它会抛出 CompletionException,因此你可以根据需要进行异常处理。

关闭虚拟线程池:

最后,通过 executorService.shutdown() 关闭线程池,释放资源。

public static void main(String[] args) throws InterruptedException {         // 创建虚拟线程的线程池         ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor();          // 假设我们有10个任务,每个任务返回一个字符串         int numTasks = 10;         List<CompletableFuture<String>> futures = new ArrayList<>(numTasks);          // 提交任务到虚拟线程池         for (int i = 0; i < numTasks; i++) {             int taskId = i;             // 将每个任务的结果放入 CompletableFuture 中             CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {                 try {                     // 模拟工作                     System.out.println("Task " + taskId + " started on " + Thread.currentThread());                     Thread.sleep(1000);  // 模拟延迟                     String result = "Result of task " + taskId;                     System.out.println("Task " + taskId + " completed on " + Thread.currentThread());                     return result;                 } catch (InterruptedException e) {                     Thread.currentThread().interrupt();                     return "Task " + taskId + " was interrupted";                 }             }, executorService);              futures.add(future);  // 将每个 future 加入集合         }          // 等待所有任务完成并获取结果         CompletableFuture<Void> allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));         allOf.join();  // 阻塞直到所有任务完成          // 合并所有任务的结果到一个集合中         List<String> results = futures.stream()                                       .map(CompletableFuture::join)  // 获取每个任务的结果                                       .collect(Collectors.toList());  // 合并到列表          // 打印结果         System.out.println("All results: " + results);          // 关闭虚拟线程池         executorService.shutdown();     } 

Java 不确定线程数,要异步多线程执行,还要等待所有线程执行结束,然后获取结果合并

解释:

任务列表 (tasks):我们创建了一个 List<Callable> 来保存所有要执行的异步任务,每个任务返回一个 Integer 结果。

创建线程池:使用 Executors.newFixedThreadPool(5) 创建了一个大小为 5 的线程池,可以并发执行 5 个线程。线程池的大小可以根据实际需要动态调整。

提交任务并获取 Future 列表:executorService.invokeAll(tasks) 方法会提交所有任务,并返回一个 List<Future>。每个 Future 对象代表一个异步任务的结果。

等待任务完成并合并结果:通过 future.get() 方法阻塞当前线程,直到任务完成并返回结果。我们在 sum 中累加所有任务的结果。

关闭线程池:最后,使用 executorService.shutdown() 关闭线程池,确保所有线程在任务完成后能够被正确回收。

重要事项

  • invokeAll():会阻塞当前线程,直到所有任务完成。如果任务执行的时间不确定,使用 invokeAll() 是比较合适的,它会等待所有任务完成,并返回 Future 列表。
  • Future.get():该方法会阻塞当前线程,直到任务完成。如果任务执行有异常,get() 会抛出异常。
  • 线程池管理:使用 ExecutorService 方便管理线程池的大小,避免频繁创建和销毁线程带来的性能损失。
public static void main(String[] args) throws InterruptedException, ExecutionException {         // 假设我们有一些任务需要并发执行         List<Callable<Integer>> tasks = new ArrayList<>();                  // 创建一些任务         for (int i = 0; i < 10; i++) {             final int taskId = i;             tasks.add(() -> {                 // 模拟任务执行,返回一个结果                 Thread.sleep(1000);  // 模拟任务耗时                 return taskId * 2;   // 假设任务返回 taskId 的 2 倍             });         }          // 创建一个固定大小的线程池         ExecutorService executorService = Executors.newFixedThreadPool(5);          try {             // 提交所有任务并返回一个 Future 列表             List<Future<Integer>> futures = executorService.invokeAll(tasks);              // 等待所有任务完成并合并结果             int sum = 0;             for (Future<Integer> future : futures) {                 sum += future.get();  // 获取任务结果并合并             }              // 输出所有任务的合并结果             System.out.println("Total sum: " + sum);          } finally {             // 关闭线程池             executorService.shutdown();         }     } 

实际案例 多线程调API然后合并API的结果返回给前端

声明任务队列集合

        /*变量值对应Map*/         List<VarResultDto> results = new ArrayList<>();         // 假设我们有一些任务需要并发执行         List<Callable<Map<String, Object>>> tasks = new ArrayList<>(); 

将任务加入然后加入任务队列

   tasks.add(() -> {                     Map<String, Object> respTask = new HashMap<>();                     List<VarResultDto> listTaskResp = new ArrayList<>();                     List<String> listTaskError = new ArrayList<>();                     try {                         log.info("执行API请求{} apiId:[{}]", vo.getApiUrl(), vo.getId());                         /*请求API获取结果*/                         R<Object> objectR = apiDataInfoService.executeApi(vo);                         // 解析结果                         JSONObject apiResp = JSONUtil.parseObj(objectR);                         if (apiResp.getInt("code") == 200 || apiResp.getInt("code") == 0) {                             apiResp = apiResp.getJSONObject("data");                         }                         // JavaScript数据处理                         if (StringUtils.isNotBlank(apiVarInfoDto.getJs())) {                             try {                                 String newJson = SpringUtils.execJavaScript(JSON.toJSONString(apiResp), apiVarInfoDto.getJs());                                 apiResp = JSONUtil.parseObj(newJson);                                 log.info("JavaScript数据处理完成");                             } catch (Exception e) {                                 log.warn("JavaScript数据处理异常: {}", JSON.toJSONString(apiVarInfoDto));                             }                         }                          final JSONObject tempData = apiResp;                         relations.forEach(relation -> {                             String value = JSONUtil.getByPath(tempData, relation.getResult(), "");                             if (StringUtils.isNotBlank(value)) {                                 // *设置变量及实际值*                                 VarResultDto resultDto = new VarResultDto();                                 resultDto.setId(relation.getId());                                 resultDto.setName(relation.getName());                                 resultDto.setResult(value);                                 listTaskResp.add(resultDto);                             } else {                                 String error = "API接口:[" + vo.getApiName() + "]无法取得变量:[" + relation.getName() + "]有效数据,原因:[" + "API地址:" + apiDataInfo.getApiUrl() + "->返回错误:" + tempData.toString() + "]";                                 listTaskError.add(error);                             }                         });                         respTask.put("results", listTaskResp);                         respTask.put("errorLogs", listTaskError);                     } catch (Exception e) {                         log.error("请求API->{}失败!{}", vo.getApiUrl(), e.getMessage(), e);                         boolean contains = e.getMessage().contains("TIMEOUT");                         /*记录错误结果*/                         relations.forEach(relation -> {                             String error = "API接口:[" + vo.getApiName() + "]无法取得变量:[" + relation.getName() + "]有效数据,原因:[" + (contains ? "数据接口获取超时" : e.getMessage()) + "]";                             listTaskError.add(error);                         });                         respTask.put("errorLogs", listTaskError);                     }                     return respTask;                 }); 

提交任务去执行,获取所有任务的结果,合并结果

 String defaultThreadPoolSize = configService.getConfigValue("api_fork_join_size", "5");         // 创建一个固定大小的线程池         try (ExecutorService executorService = Executors.newFixedThreadPool(Integer.parseInt(defaultThreadPoolSize))) {             try {                 // 提交所有任务并返回一个 Future 列表                 List<Future<Map<String, Object>>> futures = executorService.invokeAll(tasks);                  // 等待所有任务完成并合并结果                 List<Map<String, Object>> sum = new ArrayList<>();                 for (Future<Map<String, Object>> future : futures) {                     // 获取任务结果并合并                     sum.add(future.get());                 }                 // 输出所有任务的合并结果                 for (Map<String, Object> stringObjectMap : sum) {                     Object results1 = stringObjectMap.get("results");                     if (results1 != null) {                         results.addAll((List<VarResultDto>) results1);                     }                     Object errorLogs1 = stringObjectMap.get("errorLogs");                     if (errorLogs1 != null) {                         errorLogs.addAll((List<String>) errorLogs1);                     }                 }             } catch (Exception e) {                 log.error("多线程---并行处理--出错了{}", e.getMessage(), e);             } finally {                 // 关闭线程池                 executorService.shutdown();             }         } 

发表评论

评论已关闭。

相关文章