一、背景
在实际开发中，有些业务需要依赖多线程的返回数据，不是单纯的只执行业务逻辑就好。
二、demo
1.线程池工具类
@Slf4j
public class ThreadPoolUtils {private static ThreadPoolTaskExecutor poolTaskExecutor = null;private ThreadPoolUtils() {}public static ThreadPoolTaskExecutor getInstance() {try {if (poolTaskExecutor == null) {poolTaskExecutor = new ThreadPoolTaskExecutor();// 设置核心线程数poolTaskExecutor.setCorePoolSize(10);// 设置最大线程数poolTaskExecutor.setMaxPoolSize(100);// 设置队列容量poolTaskExecutor.setQueueCapacity(50000);// 设置线程活跃时间（秒）poolTaskExecutor.setKeepAliveSeconds(200);// 设置默认线程名称poolTaskExecutor.setThreadNamePrefix("taskExecutor-");// 设置拒绝策略 不在新线程中执行任务，而是有调用者所在的线程来执行poolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());// 等待所有任务结束后再关闭线程池poolTaskExecutor.setWaitForTasksToCompleteOnShutdown(true);// 加载poolTaskExecutor.initialize();} else {log.info("ThreadPool-"+ "activeCount=" + poolTaskExecutor.getThreadPoolExecutor().getActiveCount()+ ",taskCount=" + poolTaskExecutor.getThreadPoolExecutor().getTaskCount()+ ",completedTaskCount=" + poolTaskExecutor.getThreadPoolExecutor().getCompletedTaskCount()+ ",largestPoolSize=" + poolTaskExecutor.getThreadPoolExecutor().getLargestPoolSize()+ ",queueSize=" + poolTaskExecutor.getThreadPoolExecutor().getQueue().size());}} catch (Exception e) {log.error("[线程池使用] - 发生异常！", e);}return poolTaskExecutor;}2. 并行任务工具类（范型、多线程）
@Slf4j
public class ParallelUtil {@FunctionalInterfacepublic interface ParallelFunction<T> extends Supplier<T> {}@Data@Accessors(chain = true)public static class ParallelJob<T> {private ParallelFunction<T> function;private T result;}public static void execute(@NonNull List<ParallelJob> jobs) {log.info("并行任务工具类 - 开始执行，任务数:{}", jobs.size());CountDownLatch countDownLatch = new CountDownLatch(jobs.size());for (int num = 0; num < jobs.size(); num++) {int tempNum = num + 1;ParallelJob job = jobs.get(num);ThreadPoolUtils.getInstance().execute(() -> {try {log.info("并行任务工具类 - 第{}个任务执行中..", tempNum);// 处理业务逻辑job.setResult(job.getFunction().get());} catch (Exception e) {log.info("并行任务工具类 - 第{}个任务执行失败e:{}", tempNum, e);} finally {countDownLatch.countDown();}});}try {countDownLatch.await();} catch (InterruptedException e) {e.printStackTrace();}log.info("并行任务工具类 - 执行完成！");}public static void execute(@NonNull ParallelJob... jobs) {log.info("并行任务工具类 - 开始执行，任务数:{}", jobs.length);Arrays.stream(jobs).parallel().forEach(job ->job.setResult(job.getFunction().get()));log.info("并行任务工具类 - 执行完成！");}}3.单元测试
@Testpublic void test1() {StopWatch duration = new StopWatch();duration.start();// 任务定义String mobile = "12345678";ParallelUtil.ParallelJob<List<String>> userNamesJob = new ParallelUtil.ParallelJob<List<String>>().setFunction(() -> {return queryUserNamesByMobile(mobile);});ParallelUtil.ParallelJob<Integer> userCountJob = new ParallelUtil.ParallelJob<Integer>().setFunction(() -> {return getCountNum();});List<ParallelUtil.ParallelJob> jobs = new ArrayList<>();jobs.add(userNamesJob);jobs.add(userCountJob);// 执行任务
// ParallelUtil.execute(jobs);ParallelUtil.execute(userNamesJob,userCountJob);// 结果输出System.out.printf("执行耗时：" + duration.getTime(TimeUnit.MILLISECONDS) + ":userNames:" + JSON.toJSONString(userNamesJob.getResult()) + ",userCount:" + userCountJob.getResult());}private Integer getCountNum() {try {Thread.sleep(10000L);} catch (InterruptedException e) {e.printStackTrace();}return 3;}private List<String> queryUserNamesByMobile(String mobile) {// 业务处理try {Thread.sleep(10000L);} catch (InterruptedException e) {e.printStackTrace();}return Arrays.asList("张三", "李四", "王五");}