LatchUtils:简化Java异步任务同步的利器

B站影视 韩国电影 2025-10-23 16:32 2

摘要:在Java应用开发中,为了提升系统性能和响应速度,我们经常需要将一些耗时操作(如调用外部API、查询数据库、复杂计算等)进行异步并行处理。当主流程需要等待所有这些并行任务执行完毕后再继续时,我们通常会用到 ExecutorService、 CountDownL

在Java应用开发中,为了提升系统性能和响应速度,我们经常需要将一些耗时操作(如调用外部API、查询数据库、复杂计算等)进行异步并行处理。当主流程需要等待所有这些并行任务执行完毕后再继续时,我们通常会用到 ExecutorService、 CountDownLatch 等并发工具。

然而,直接使用这些原生工具,往往意味着需要编写一些重复的、模式化的“胶水代码”,这不仅增加了代码量,也让核心业务逻辑显得不够清晰。

为了解决这个问题,我封装了一个名为 LatchUtils 的轻量级工具类。它能够以一种极其简洁的方式来组织和管理这一类异步任务。

import java.util.LinkedList;import java.util.List;import java.util.concurrent.CountDownLatch;import java.util.concurrent.Executor;import java.util.concurrent.TimeUnit;public class LatchUtils { private static final ThreadLocal> THREADLOCAL = ThreadLocal.withInitial(LinkedList::new); public static void submitTask(Executor executor, Runnable runnable) { THREADLOCAL.get.add(new TaskInfo(executor, runnable)); } private static ListtaskInfos = THREADLOCAL.get; THREADLOCAL.remove; return taskInfos; } public static boolean waitFor(long timeout, TimeUnit timeUnit) { ListtaskInfos = popTask; if (taskInfos.isEmpty) { return true; } CountDownLatch latch = new CountDownLatch(taskInfos.size); for (TaskInfo taskInfo : taskInfos) { Executor executor = taskInfo.executor; Runnable runnable = taskInfo.runnable; executor.execute( -> { try { runnable.run; } finally { latch.countDown; } }); } boolean await = false; try { await = latch.await(timeout, timeUnit); } catch (Exception ignored) { } return await; } private static final class TaskInfo { private final Executor executor; private final Runnable runnable; public TaskInfo(Executor executor, Runnable runnable) { this.executor = executor; this.runnable = runnable; } }}

LatchUtils 的设计哲学是:多次提交,一次等待。

任务注册: 在主流程代码中,可以先通过 LatchUtils.submitTask 提交Runnable任务和其对应的Executor(该线程池用来执行这个Runnable)。• 执行并等待: 当并行任务都提交完毕后,你只需调用一次 LatchUtils.waitFor。关注工众号:码猿技术专栏,回复关键词:1111 获取阿里内部Java性能调优手册!该方法会立即触发所有已注册任务的执行,并阻塞等待所有任务执行完成或超时。

这个工具类对外暴露的接口极其简单,只有两个核心静态方法:

submitTask

public static void submitTask(Executor executor, Runnable runnable)

功能: 提交一个异步任务。

参数:

executor:java.util.concurrent.Executor - 指定执行此任务的线程池。runnable:java.lang.Runnable - 需要异步执行的具体业务逻辑。

waitFor

public static boolean waitFor(long timeout, TimeUnit timeUnit)

功能: 触发所有已提交任务的执行,并同步等待它们全部完成。

参数:

timeout:long - 最长等待时间。timeUnit:java.util.concurrent.TimeUnit - 等待时间单位。true: 如果所有任务在指定时间内成功完成。false: 如果等待超时。

注意: 该方法在执行后会自动清理当前线程提交的任务列表,因此可以重复使用。

让我们来看一个典型的应用场景:一个聚合服务需要同时调用用户服务、订单服务和商品服务,拿到所有结果后再进行下一步处理。

import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.TimeUnit;public class Main { public static void main(String args) { // 1. 准备一个线程池 ExecutorService executorService = Executors.newFixedThreadPool(3); System.out.println("主流程开始,准备分发异步任务..."); // 2. 提交多个异步任务 // 任务一:获取用户信息 LatchUtils.submitTask(executorService, -> { try { System.out.println("开始获取用户信息..."); Thread.sleep(1000); // 模拟耗时 System.out.println("获取用户信息成功!"); } catch (InterruptedException e) { Thread.currentThread.interrupt; } }); // 任务二:获取订单信息 LatchUtils.submitTask(executorService, -> { try { System.out.println("开始获取订单信息..."); Thread.sleep(1500); // 模拟耗时 System.out.println("获取订单信息成功!"); } catch (InterruptedException e) { Thread.currentThread.interrupt; } }); // 任务三:获取商品信息 LatchUtils.submitTask(executorService, -> { try { System.out.println("开始获取商品信息..."); Thread.sleep(500); // 模拟耗时 System.out.println("获取商品信息成功!"); } catch (InterruptedException e) { Thread.currentThread.interrupt; } }); System.out.println("所有异步任务已提交,主线程开始等待..."); // 3. 等待所有任务完成,最长等待5秒 boolean allTasksCompleted = LatchUtils.waitFor(5, TimeUnit.SECONDS); // 4. 根据等待结果继续主流程 if (allTasksCompleted) { System.out.println("所有异步任务执行成功,主流程继续..."); } else { System.err.println("有任务执行超时,主流程中断!"); } // 5. 关闭线程池 executorService.shutdown; }}

输出结果:

主流程开始,准备分发异步任务...所有异步任务已提交,主线程开始等待...开始获取商品信息...开始获取用户信息...开始获取订单信息...获取商品信息成功!获取用户信息成功!获取订单信息成功!所有异步任务执行成功,主流程继续...

从这个例子中可以看到,业务代码变得非常清晰。我们只需要关注“提交任务”和“等待结果”这两个动作,而无需关心 CountDownLatch 的初始化、countDown 的调用以及异常处理等细节。

为了更好地理解 LatchUtils 带来的价值,让我们看看要实现与上面完全相同的功能,用传统的Java并发API需要如何编写代码。
通常有两种主流方式:使用 CountDownLatch 或使用 CompletableFuture。

方式一:直接使用 CountDownLatch

这是最经典的方式,开发者需要手动管理 CountDownLatch 的生命周期。

import java.util.concurrent.CountDownLatch;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.TimeUnit;public class ManualCountDownLatchExample { public static void main(String args) { // 1. 准备一个线程池 ExecutorService executorService = Executors.newFixedThreadPool(3); // 2. 手动初始化 CountDownLatch,数量为任务数 CountDownLatch latch = new CountDownLatch(3); System.out.println("主流程开始,准备分发异步任务..."); // 3. 提交任务,并在每个任务的 finally 块中手动调用 latch.countDown // 任务一:获取用户信息 executorService.execute( -> { try { System.out.println("开始获取用户信息..."); Thread.sleep(1000); System.out.println("获取用户信息成功!"); } catch (InterruptedException e) { Thread.currentThread.interrupt; } finally { latch.countDown; // 手动减一 } }); // 任务二:获取订单信息 executorService.execute( -> { try { System.out.println("开始获取订单信息..."); Thread.sleep(1500); System.out.println("获取订单信息成功!"); } catch (InterruptedException e) { Thread.currentThread.interrupt; } finally { latch.countDown; // 手动减一 } }); // 任务三:获取商品信息 executorService.execute( -> { try { System.out.println("开始获取商品信息..."); Thread.sleep(500); System.out.println("获取商品信息成功!"); } catch (InterruptedException e) { Thread.currentThread.interrupt; } finally { latch.countDown; // 手动减一 } }); System.out.println("所有异步任务已提交,主线程开始等待..."); // 4. 手动调用 latch.await 进行等待 boolean allTasksCompleted = false; try { allTasksCompleted = latch.await(5, TimeUnit.SECONDS); } catch (InterruptedException e) { // 需要处理中断异常 Thread.currentThread.interrupt; System.err.println("主线程在等待时被中断!"); } // 5. 根据等待结果继续主流程 if (allTasksCompleted) { System.out.println("所有异步任务执行成功,主流程继续..."); } else { System.err.println("有任务执行超时,主流程中断!"); } // 6. 关闭线程池 executorService.shutdown; }}import java.util.concurrent.CompletableFuture;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.TimeUnit;public class CompletableFutureExample { public static void main(String args) { // 1. 准备一个线程池 ExecutorService executorService = Executors.newFixedThreadPool(3); System.out.println("主流程开始,准备分发异步任务..."); // 2. 创建 CompletableFuture 任务 CompletableFutureuserFuture = CompletableFuture.runAsync( -> { try { System.out.println("开始获取用户信息..."); Thread.sleep(1000); System.out.println("获取用户信息成功!"); } catch (InterruptedException e) { Thread.currentThread.interrupt; } }, executorService); CompletableFutureorderFuture = CompletableFuture.runAsync( -> { try { System.out.println("开始获取订单信息..."); Thread.sleep(1500); System.out.println("获取订单信息成功!"); } catch (InterruptedException e) { Thread.currentThread.interrupt; } }, executorService); CompletableFutureproductFuture = CompletableFuture.runAsync( -> { try { System.out.println("开始获取商品信息..."); Thread.sleep(500); System.out.println("获取商品信息成功!"); } catch (InterruptedException e) { Thread.currentThread.interrupt; } }, executorService); System.out.println("所有异步任务已提交,主线程开始等待..."); // 3. 使用 CompletableFuture.allOf 将所有任务组合起来 CompletableFutureallFutures = CompletableFuture.allOf(userFuture, orderFuture, productFuture); // 4. 等待组合后的 Future 完成 try { allFutures.get(5, TimeUnit.SECONDS); System.out.println("所有异步任务执行成功,主流程继续..."); } catch (Exception e) { // 需要处理多种异常,如 InterruptedException, ExecutionException, TimeoutException System.err.println("任务执行超时或出错,主流程中断! " + e.getMessage); } // 5. 关闭线程池 executorService.shutdown; }}特性LatchUtils手动CountDownLatchCompletableFuture.allOf代码简洁性极高。业务逻辑和并发控制分离,核心代码清晰。中等。需要在每个任务中嵌入latch.countDown,分散了关注点。较高。链式调用风格,但需要创建多个Future对象。状态管理自动。工具类内部自动管理CountDownLatch。手动。需要自己创建、维护和传递CountDownLatch实例。自动。由CompletableFuture框架管理任务状态。错误处理简化。waitFor内部处理InterruptedException,仅返回布尔值。复杂。需要显式地在finally中countDown,并为主线程的await处理InterruptedException。复杂。get方法会抛出多种受检异常,需要统一处理。关注点分离优秀。开发者只需关注“提交”和“等待”两个动作。一般。并发控制逻辑(countDown)侵入到了业务Runnable中。良好。任务的定义和组合是分开的,但仍需处理组合后的Future。易用性非常简单。几乎没有学习成本。需要理解CountDownLatch。容易忘记countDown或错误处理。需要理解CompletableFuture。API较为丰富,有一定学习曲线。

对于“分发一组并行任务,然后等待它们全部完成”这一特定但常见的模式,LatchUtils 通过适度的封装,极大地简化了开发者的工作。

它隐藏了并发控制的复杂性,让业务代码回归其本质,从而提高了代码的可读性和可维护性。

来源:不秃头程序员

相关推荐