本文主要是介绍Java使用多线程处理未知任务数的方案介绍，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
《Java使用多线程处理未知任务数的方案介绍》这篇文章主要为大家详细介绍了Java如何使用多线程实现处理未知任务数,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下...
知道任务个数，你可以定义好线程数规则，生成线程数去跑
代码说明：
1.虚拟线程池：
使用 Executors.newVirtualThreadPerTaskExecutor() 创建虚拟线程池，每个任务将分配一个虚拟线程来执行。
2.提交任务并返回结果：
- 每个任务通过 CompletableFuture.supplyAsync() 提交，任务会返回一个结果（这里是字符串，模拟了任务的处理结果）。
- 每个 CompletableFuture 都会保存任务的返回值。
3.等待所有任务完成：
使用 CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])) 等待所有的 CompletableFuture 完成。allOf.join() 会阻塞当前线程，直到所有任务完成。
4.收集结果：
- 使用 Java 8 的 stream() 方法和 Collectors.toList() 来收集所有任务的结果，并将它们合并到一个 List 中。
- CompletableFuture::join 会获取每个任务的结果，并且如果任务有异常，它会抛出 CompletionException，因此你可以根据需要进行异常处理。
5.关闭虚拟线程池：
最后，通过 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 (IChina编程nterruptedException e) {
Thread.currentThread().interrupt();
return "Task " + taskId + " was interrupted";
}
}, executorService);
futures.add(future); // 将每个 future 加入集合
}
// 等待所有任务完成并获取结果
CompletableFuture<Voiwww.chinasem.cnd> 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的结果返回给前端
1.声明任务队列集合
/*变量值对应Map*/
List<VarResultDto> results = new ArrayList<>();
// 假设我们有一些任务需要并发执行
List<Callable<Map<String, Object>>> tasks = new ArrayList<>();
2.将任务加入然后加入任务队列
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(apiVarInfoChina编程Dto));
}
}
final JSONObject tempData = apiResp;
relations.forEach(relation -> {
String value = JSONUtil.getByPath(tempDatphpa, 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;
});
3.提交任务去执行，获取所有任务的结果，合并结果
String defaultThreadPoolSize = configService.getConfigValue("api_fork_join_size", "5");
// 创建一个固定大小的线程池
try (ExecutorService executorService = Executors.newFixedThreadPool(Integer.parseInt(defaultThreadPoolSize))) {
try {
// 提交所有任务并返回一个 Future 列表
List<Future<Map<Strinphpg, 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();
}
}到此这篇关于Java使用多线程处理未知任务数的方案介绍的文章就介绍到这了,更多相关Java多线程处理未知任务数内容请搜索China编程(www.chinasem.cn)以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程China编程(www.chinasem.cn)！
这篇关于Java使用多线程处理未知任务数的方案介绍的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
