如何自定义Java 21结构化并发API的任务终止策略？

业务需求 #

从多个数据源获取设备的多项错误状态信息，需满足：

• 只要任一子任务返回true，立即终止所有剩余子任务，主任务返回true
• 若所有子任务返回false或空值，主任务在全部子任务完成后返回false

现有代码 #

用户当前的实现框架：

try (var scope = new StructuredTaskScope<>()) {
    scope.fork(() -> posRepo.findInspectionByUnitId(unitId));
    scope.fork(() -> posRepo.findEventOfFireByUnitId(unitId));
    scope.fork(() -> faultRepo.findGeneralErrorByUnitId(unitId));
    scope.fork(() -> ardInfRepo.findArdModeByUnitId(unitId));
    scope.fork(() -> ardInfRepo.findArdPowerStatusByUnitId(unitId));
    scope.fork(() -> ardInfRepo.findArdModeByUnitId(unitId));
    // Once a task return true, close all remaining tasks,and return the signal:"true" to main task;
    scope.join();
} catch (InterruptedException e) {
    throw new RuntimeException(e);
}

仓储接口定义：

public interface ElePosRepo extends CrudRepository<PositionInformation,Integer> {
    Optional<Boolean> findInspectionByUnitId(Integer unitId);
    Optional<Boolean> findEventOfFireByUnitId(Integer unitId);
}

解决方案 #

利用Java 19+引入的StructuredTaskScope.ShutdownOnSuccess特性，它会在第一个任务返回预期结果时自动关闭所有其他子任务，完美匹配快速失败的需求。

完整实现代码 #

public boolean checkDeviceErrorStatus(Integer unitId) {
    try (var scope = new StructuredTaskScope.ShutdownOnSuccess<Boolean>()) {
        // 将每个Optional<Boolean>转换为Boolean，空值视为false
        scope.fork(() -> posRepo.findInspectionByUnitId(unitId).orElse(false));
        scope.fork(() -> posRepo.findEventOfFireByUnitId(unitId).orElse(false));
        scope.fork(() -> faultRepo.findGeneralErrorByUnitId(unitId).orElse(false));
        scope.fork(() -> ardInfRepo.findArdModeByUnitId(unitId).orElse(false));
        scope.fork(() -> ardInfRepo.findArdPowerStatusByUnitId(unitId).orElse(false));
        // 注意：重复的任务可以考虑移除，避免冗余查询
        // scope.fork(() -> ardInfRepo.findArdModeByUnitId(unitId).orElse(false));

        // 等待第一个成功结果（即返回true），或所有任务完成
        scope.join();
        
        // 如果存在成功结果（true），返回true；否则返回false
        return scope.result().orElse(false);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new RuntimeException("设备状态检查被中断", e);
    } catch (ExecutionException e) {
        // 处理子任务抛出的异常，这里可以根据业务需求选择返回true或false，或抛出异常
        throw new RuntimeException("子任务执行失败", e.getCause());
    }
}

关键说明 #

1. ShutdownOnSuccess作用：当任何一个fork的任务返回true时，该scope会立即关闭所有未完成的子任务，终止它们的执行，避免不必要的资源消耗。
2. Optional转换：通过.orElse(false)把Optional<Boolean>的空值统一转换成false，确保只有明确的true才会触发快速终止。
3. 异常处理：
   • InterruptedException：恢复线程中断状态，避免中断信号丢失
   • ExecutionException：捕获子任务抛出的异常，根据业务需求处理（比如视为错误状态返回true，或抛出异常）
4. 冗余任务优化：代码中重复的findArdModeByUnitId任务可以移除，减少重复查询。

内容的提问来源于stack exchange，提问作者liang li