我要评论知道任务个数，你可以定义好线程数规则，生成线程数去跑
代码说明：
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 (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的结果返回给前端
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(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;
});
3.提交任务去执行，获取所有任务的结果，合并结果
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();
}
}到此这篇关于java使用多线程处理未知任务数的方案介绍的文章就介绍到这了,更多相关java多线程处理未知任务数内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！
发表评论