并发框架执行多任务的小实践

框架执行多任务的小实践"/>

并发框架执行多任务的小实践

并发框架定义

public class FutureTaskWorker {# 查看处理器数private static final int processors = Runtime.getRuntime().availableProcessors();# 定义线程池private static final ExecutorService executor = Executors.newFixedThreadPool(processors + 1);static {# 查看一下机器处理器数System.out.println(processors + " processors");}# 实现方式1----------------------------------------------------------------------------public static <T, R> List<R> handle(List<T> tasks, Function<T, R> function) {List<CompletableFuture<R>> featureRes = new ArrayList<>();for (T task : tasks) {# 任务添加到线程池CompletableFuture<R> feature = CompletableFuture.supplyAsync(() -> function.apply(task), executor);featureRes.add(feature);}List<R> res = new ArrayList<>();try {# 等所有任务执行完成CompletableFuture.allOf(featureRes.toArray(new CompletableFuture[0])).get();# 收集执行结果for (CompletableFuture<R> feature : featureRes) {res.add(feature.get());}return res;} catch (InterruptedException | ExecutionException e) {throw new RuntimeException(e);}}# 实现方式2----------------------------------------------------------------------------public static <T, R> List<R> handle1(List<T> tasks, Function<T, R> function) {# 任务添加到线程池List<CompletableFuture<R>> futureList = tasks.stream().map(task -> CompletableFuture.supplyAsync(() -> function.apply(task), executor)).collect(Collectors.toList());# 等所有任务执行完成CompletableFuture<Void> allCompletableFuture = CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0]));# 收集执行结果return allCompletableFuture.thenApply(e -> futureList.stream().map(CompletableFuture::join).collect(Collectors.toList())).join();}public static void shunDown() {executor.shutdownNow();}
}

并发框架使用

public static void main(String[] args) {List<String> names = new ArrayList<>();names.add("小明1");names.add("小明2");names.add("小明3");names.add("小明4");names.add("小明5");names.add("小明6");names.add("小明7");names.add("小明8");names.add("小明9");try {# 使用并发框架来执行多任务List<String> res = handle1(names, name -> businessMethod(name));res.forEach(System.out::println);} finally {# 关闭线程池shunDown();}
}# 这里模拟业务方法
public static String businessMethod(String str) {System.out.println(Thread.currentThread().getName() + "线程：" + new Date());return str + "  " + new Date();
}# 执行结果
16 processors
pool-1-thread-1线程：Fri Nov 03 15:18:10 CST 2023
pool-1-thread-9线程：Fri Nov 03 15:18:10 CST 2023
pool-1-thread-6线程：Fri Nov 03 15:18:10 CST 2023
pool-1-thread-5线程：Fri Nov 03 15:18:10 CST 2023
pool-1-thread-3线程：Fri Nov 03 15:18:10 CST 2023
pool-1-thread-2线程：Fri Nov 03 15:18:10 CST 2023
pool-1-thread-7线程：Fri Nov 03 15:18:10 CST 2023
pool-1-thread-8线程：Fri Nov 03 15:18:10 CST 2023
pool-1-thread-4线程：Fri Nov 03 15:18:10 CST 2023
小明1  Fri Nov 03 15:18:10 CST 2023
小明2  Fri Nov 03 15:18:10 CST 2023
小明3  Fri Nov 03 15:18:10 CST 2023
小明4  Fri Nov 03 15:18:10 CST 2023
小明5  Fri Nov 03 15:18:10 CST 2023
小明6  Fri Nov 03 15:18:10 CST 2023
小明7  Fri Nov 03 15:18:10 CST 2023
小明8  Fri Nov 03 15:18:10 CST 2023
小明9  Fri Nov 03 15:18:10 CST 2023

参考：

使用CompletableFuture.allOf实现异步执行同步搜集结果
CompletableFuture用法详解