如何在Spring Boot JPA中并行执行Delete查询并保证事务性

问题根源 #

你的代码抛出TransactionRequiredException的核心原因是：Spring声明式事务通过ThreadLocal绑定在当前线程，而CompletableFuture.runAsync()默认使用ForkJoinPool创建新线程，新线程无法继承主线程的事务上下文，导致Repository操作找不到有效事务。此外，JPA的EntityManager并非线程安全，不能跨线程共享同一个事务实例，直接跨线程复用事务不可行。

可行方案 #

方案一：合并删除操作（推荐） #

将三个删除操作合并为一个数据库批量操作，利用数据库的原子性保证事务一致性，同时减少IO交互开销，性能比并行更优。

实现步骤：

1. 在任意一个Repository中添加批量删除方法（以FlowPlanDetailsRepository为例）：

@Modifying
@Transactional
@Query(nativeQuery = true, value = """
    DELETE FROM wavemaker_flow_plan WHERE flow_plan_id = ?1;
    DELETE FROM wavemaker_idc_calculations WHERE flow_plan_id = ?1;
    DELETE FROM flow_plan_details WHERE flow_plan_id = ?1;
    """)
void batchDeleteByFlowPlanId(Long flowPlanId);

2. 在原方法中替换三个独立的删除调用：

// 替换原来的三个Repository.deleteByFlowPlanId调用
flowPlanDetailsRepository.batchDeleteByFlowPlanId(flowPlanId);

注意：需确保数据库支持多语句执行（如MySQL需在JDBC连接URL中添加allowMultiQueries=true）。数据库会保证这个批量操作的原子性，任意一条删除失败，整个操作都会回滚。

方案二：分布式事务实现并行删除（适合必须并行的场景） #

如果业务必须并行执行三个删除操作，需要通过分布式事务（XA事务）保证跨线程的事务原子性。

步骤1：配置Spring异步线程池

@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {
    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(3);
        executor.setMaxPoolSize(3);
        executor.setQueueCapacity(10);
        executor.setThreadNamePrefix("DeleteFlowPlan-");
        executor.initialize();
        return executor;
    }
}

步骤2：封装异步删除服务

将每个删除操作为独立的异步方法，并绑定到XA事务管理器：

@Service
public class FlowPlanDeleteService {
    private final WavemakerFlowPlanRepository wavemakerFlowPlanRepository;
    private final WavemakerIdcCalculationsRepository wavemakerIdcCalculationsRepository;
    private final FlowPlanDetailsRepository flowPlanDetailsRepository;

    public FlowPlanDeleteService(WavemakerFlowPlanRepository wavemakerFlowPlanRepository,
                                 WavemakerIdcCalculationsRepository wavemakerIdcCalculationsRepository,
                                 FlowPlanDetailsRepository flowPlanDetailsRepository) {
        this.wavemakerFlowPlanRepository = wavemakerFlowPlanRepository;
        this.wavemakerIdcCalculationsRepository = wavemakerIdcCalculationsRepository;
        this.flowPlanDetailsRepository = flowPlanDetailsRepository;
    }

    @Async
    @Transactional(transactionManager = "xaTransactionManager", propagation = Propagation.REQUIRED)
    public CompletableFuture<Void> deleteWavemakerFlowPlan(Long flowPlanId) {
        wavemakerFlowPlanRepository.deleteByFlowPlanId(flowPlanId);
        return CompletableFuture.completedFuture(null);
    }

    @Async
    @Transactional(transactionManager = "xaTransactionManager", propagation = Propagation.REQUIRED)
    public CompletableFuture<Void> deleteWavemakerIdcCalculations(Long flowPlanId) {
        wavemakerIdcCalculationsRepository.deleteByFlowPlanId(flowPlanId);
        return CompletableFuture.completedFuture(null);
    }

    @Async
    @Transactional(transactionManager = "xaTransactionManager", propagation = Propagation.REQUIRED)
    public CompletableFuture<Void> deleteFlowPlanDetails(Long flowPlanId) {
        flowPlanDetailsRepository.deleteByFlowPlanId(flowPlanId);
        return CompletableFuture.completedFuture(null);
    }
}

步骤3：在原方法中调用异步服务

@Transactional(transactionManager = "xaTransactionManager")
public ServiceResponse<Void> deleteFlowPlan(Context context) {
    // 原参数校验和查询逻辑保持不变
    ServiceRequest<Map<String, Object>> serviceRequest = context.get(BaseKeys.DELETE_FLOW);
    Map<String,Object> request = serviceRequest.getPayload();
    Long flowPlanId= (Long) request.get("flowPlanId");

    if(Objects.isNull(flowPlanId))
        throw new AllocationRuntimeException(ErrorCodes.INVALID_INPUT.getError("Flowplanid can not be null."));

    FlowPlanDetails flowPlanDetails = flowPlanDetailsRepository.findByFlowPlanId(flowPlanId);
    if(Objects.isNull(flowPlanDetails))
        throw new AllocationRuntimeException(ErrorCodes.INVALID_INPUT.getError("Flowplanid could not be found."));

    // 触发异步删除
    CompletableFuture<Void> future1 = flowPlanDeleteService.deleteWavemakerFlowPlan(flowPlanId);
    CompletableFuture<Void> future2 = flowPlanDeleteService.deleteWavemakerIdcCalculations(flowPlanId);
    CompletableFuture<Void> future3 = flowPlanDeleteService.deleteFlowPlanDetails(flowPlanId);

    // 等待所有操作完成，任意失败则触发回滚
    try {
        CompletableFuture.allOf(future1, future2, future3).get();
    } catch (InterruptedException | ExecutionException e) {
        throw new AllocationRuntimeException(ErrorCodes.OPERATION_FAILED.getError("Delete flow plan failed."), e.getCause());
    }

    return ServiceResponseBuilder.get(null);
}

注意：分布式事务会增加系统复杂度和性能开销，仅在业务必须并行的场景下使用，需提前配置好XA事务管理器（如Atomikos）及支持XA的数据源。

内容的提问来源于stack exchange，提问作者Elvis Price