当前位置：首页 > news >正文

SpringBoot的并行SQL任务并完成所有任务之后返回操作

news 来源：原创 2025/5/15 3:12:46

一、核心实现方案

1. 线程池配置与异步支持

通过 @EnableAsync 启用异步支持，并自定义线程池避免默认线程池的性能问题：

@Configuration
@EnableAsync
public class AsyncConfig {
    @Bean
    public Executor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);    // 根据业务负载调整
        executor.setMaxPoolSize(10);    // 突发流量缓冲
        executor.setQueueCapacity(100); // 防止 OOM
        executor.setThreadNamePrefix("DB-Async-");
        executor.initialize();
        return executor;
    }
}

关键点：
• 线程池隔离避免资源竞争（如 IO 密集型与 CPU 密集型任务分开）
• 拒绝策略需根据业务场景定制（如日志记录或降级处理）

2. 服务层异步方法封装

通过 @Async 注解将数据库查询封装为异步任务：

@Service
public class AsyncQueryService {
    @Autowired
    private UserRepository userRepo;
    
    @Async
    public CompletableFuture<List<User>> queryUsers() {
        return CompletableFuture.supplyAsync(userRepo::findAll);
    }

    @Async
    public CompletableFuture<List<Order>> queryOrders() {
        return CompletableFuture.supplyAsync(orderRepo::findAll);
    }
}

关键点：
• @Async 方法需定义在独立 @Service 类中（Spring 代理机制限制）
• 返回 CompletableFuture 以实现链式编排

3. 控制层任务编排与结果合并

使用 CompletableFuture.allOf() 实现任务同步等待，并通过 join() 提取结果：

@RestController
public class DataController {
    @Autowired
    private AsyncQueryService asyncService;

    @GetMapping("/parallel-query")
    public ResponseEntity<Map<String, Object>> parallelQuery() {
        CompletableFuture<List<User>> usersFuture = asyncService.queryUsers();
        CompletableFuture<List<Order>> ordersFuture = asyncService.queryOrders();

        CompletableFuture.allOf(usersFuture, ordersFuture).join(); // 阻塞等待

        Map<String, Object> result = new HashMap<>();
        result.put("users", usersFuture.join());
        result.put("orders", ordersFuture.join());
        return ResponseEntity.ok(result);
    }
}

关键点：
• allOf().join() 会阻塞主线程直至所有任务完成
• 生产环境需添加超时控制（如 orTimeout(5, SECONDS)）

二、高级优化策略

1. 动态数据源管理

若查询涉及多个数据库（如 MySQL 和 SQL Server），需配置多数据源：

# application.yml
spring:
  datasource:
    dynamic:
      primary: mysql
      datasource:
        mysql:
          url: jdbc:mysql://localhost:3306/db1
        sqlserver:
          url: jdbc:sqlserver://localhost:1433;databaseName=db2

结合 ThreadLocal 和 AbstractRoutingDataSource 实现动态切换：

public class DynamicDataSource extends AbstractRoutingDataSource {
    @Override
    protected Object determineCurrentLookupKey() {
        return DataSourceContextHolder.getDataSource(); // 从ThreadLocal获取数据源标识
    }
}

2. 结果合并与链式编程

使用 thenCombine() 优化结果合并逻辑：

CompletableFuture<CombinedResult> combinedFuture = usersFuture
    .thenCombine(ordersFuture, (users, orders) -> new CombinedResult(users, orders));

优势：
• 避免嵌套回调，提升代码可读性
• 支持异常传递与统一处理

3. 事务与一致性保障

在异步任务中管理事务需注意：
• 默认 @Async 方法不在事务上下文中执行，需显式配置 @Transactional
• 多数据源场景需为每个数据源单独配置事务管理器

三、性能与异常处理

1. 线程池监控

通过 ThreadPoolTaskExecutor 方法监控：

int activeCount = executor.getActiveCount(); // 活跃线程数
int queueSize = executor.getQueue().size();  // 队列堆积情况

2. 异常兜底策略

usersFuture.exceptionally(ex -> {
    log.error("用户查询失败", ex);
    return Collections.emptyList(); // 返回降级结果
});

四、适用场景对比

方案	适用场景	优点	限制
`@Async` + `CompletableFuture`	简单查询并行化	代码简洁，Spring 原生支持	事务管理复杂
`ExecutorService`	需精细控制线程池或非 Spring 环境	灵活性高	需手动管理生命周期
JDBC 异步 API	低延迟、高吞吐场景	无框架依赖	需数据库驱动支持