Spring Boot并发场景下异步超时重置为30秒问题求助

问题描述 #

在Spring Boot开发GraphQL resolver实现批量数据库查询时，串行处理请求时5分钟超时配置正常生效，但引入flatMap、parallel()、Schedulers等并发/异步逻辑后，请求会在30秒后超时，尽管已通过多组配置将超时设为5分钟。

相关代码片段 #

@SchemaMapping(typeName = "Items", field = "somePricing")
public Mono<SomePricingResponse> getSomePricing(
       @Argument SomePricingRequest request,
       @Argument Integer pageNo,
       @Argument Integer pageSize) {
   RequestParams params = inputValidator.constructAndValidateRequestParams(pageNo, pageSize);
   inputValidator.validateSomePricingRequest(request);
   return pricingService.getSomePricing(request, params)
           .doOnSuccess(response -> LoggingUtil.logInfo("Fetched Some Pricing"));
}

private Mono<SomePricingResponse> handleBatchingForSomeCombinations(SomePricingRequest request, RequestParams params) {
   List<SomeCombination> allCombinations = request.getSomeCombinations();
   int batchSize = 200;

   // Split the combinations into batches
   List<List<SomeCombination>> batches = new ArrayList<>();
   for (int i = 0; i < allCombinations.size(); i += batchSize) {
       batches.add(allCombinations.subList(i, Math.min(i + batchSize, allCombinations.size())));
   }

   LoggingUtil.logInfo("Total number of batches: {}", batches.size());

   return Flux.fromIterable(batches)
           .timeout(Duration.ofMinutes(5))
           .parallel(10)
           .runOn(Schedulers.newParallel("batch-processing", 10))
           .flatMap(batch -> {
               SomePricingRequest batchRequest = new SomePricingRequest();
               batchRequest.setSomeCombinations(batch);

               int batchNum = batches.indexOf(batch) + 1;
               LoggingUtil.logInfo("Processing batch number {} with {} combinations", batchNum, batch.size());

               // Execute the batch query without a timeout
               return pricingDataAccessLayer.getSomePricing(batchRequest, params)
                       .doOnSuccess(response -> LoggingUtil.logInfo("Batch number {} processed successfully", batchNum));
           }) // Control concurrency here
           .sequential()
           .collectList()
           .map(responses -> {
               SomePricingResponse finalResponse = new SomePricingResponse();
               finalResponse.setData(new ArrayList<>());

               for (SomePricingResponse response : responses) {
                   if (response != null && response.getData() != null) {
                       finalResponse.getData().addAll(response.getData());
                   } else {
                       LoggingUtil.logWarn("One of the batch responses was null or contained null data");
                   }
               }

               finalResponse.setPageable(new Pageable((long) finalResponse.getData().size(), 1L, finalResponse.getData().size(), 0));
               LoggingUtil.logInfo("All batches processed successfully with a total of {} records", finalResponse.getData().size());
               return finalResponse;
           });
}

已尝试的配置 #

spring:
  mvc:
    async:
      request-timeout: 600000  # 10 minutes
server:
  tomcat:
    connection-timeout: 600000  # 10 minutes
    async-timeout: 600000  # 10 minutes
  async:
    request-timeout: 300000  # 5 minutes
  servlet:
    async:
      timeout: 300000  # 5 minutes
graphql:
  servlet:
    asyncTimeout: 600000

超时日志信息 #

2024-08-15 14:22:17.044 [DEBUG] [nio-8080-exec-9] o.s.w.c.request.async.WebAsyncManager: Started async request for "/aais-platform-gql/graphql"
2024-08-15 14:22:17.045 [DEBUG] [nio-8080-exec-9] o.s.web.servlet.DispatcherServlet: Exiting but response remains open for further handling
2024-08-15 14:22:47.153 [DEBUG] [io-8080-exec-10] o.s.w.c.request.async.WebAsyncManager: Servlet container timeout notification for "/aais-platform-gql/graphql"
2024-08-15 14:22:47.154 [DEBUG] [io-8080-exec-10] o.s.w.c.request.async.WebAsyncManager: Async result set to: org.springframework.web.context.request.async.AsyncRequestTimeoutException for "/aais-platform-gql/graphql"

需求指导 #

• 排查是否存在隐藏配置或外部服务覆盖了超时设置；
• 可绕过30秒限制的并发处理替代方案；
• 分析应用与Tomcat服务器或Spring异步处理的交互是否存在问题。

问题分析与解决方案 #

1. 隐藏配置/外部覆盖排查

• 验证配置实际生效值：启用Spring Boot Actuator，访问/actuator/configprops查看server.servlet.async.timeout等配置的实际生效值，确认是否被其他配置覆盖。
• 检查代码层面覆盖：查看是否存在WebMvcConfigurer实现类重写configureAsyncSupport方法手动设置超时，或全局过滤器/拦截器修改AsyncWebRequest的超时参数。
• 外部Tomcat配置检查：若使用外部Tomcat，检查server.xml中Connector节点的asyncTimeout属性，该配置会覆盖application.yml中的Tomcat超时设置。
• GraphQL版本适配：确认graphql-spring-boot-starter版本，部分旧版本中graphql.servlet.asyncTimeout配置键可能为graphql.servlet.async-timeout。

2. 并发处理替代方案

• 使用Spring管理线程池：替换Schedulers.newParallel为Spring提供的TaskScheduler，确保异步任务在Spring上下文内执行：

  @Autowired
  private TaskScheduler taskScheduler;
  
  // 修改runOn逻辑
  return Flux.fromIterable(batches)
          .timeout(Duration.ofMinutes(5))
          .parallel(10)
          .runOn(Schedulers.from(taskScheduler))
          // 后续逻辑保持不变
  

• 用flatMap并发参数替代parallel：直接通过flatMap的第二个参数控制并发数，无需切换线程池：

  return Flux.fromIterable(batches)
          .timeout(Duration.ofMinutes(5))
          .flatMap(batch -> {
              // 批量处理逻辑
          }, 10) // 控制并发数为10
          .collectList()
          // 后续映射逻辑保持不变
  

• 手动绑定请求上下文：若必须用自定义线程池，需手动绑定请求上下文：

  RequestAttributes attributes = RequestContextHolder.getRequestAttributes();
  return pricingDataAccessLayer.getSomePricing(batchRequest, params)
          .subscribeOn(Schedulers.newParallel("batch-processing", 10))
          .doOnSubscribe(s -> RequestContextHolder.setRequestAttributes(attributes))
          .doOnTerminate(() -> RequestContextHolder.resetRequestAttributes());
  

3. 应用与Tomcat/Spring异步交互分析

• 上下文丢失导致超时失效：Spring的WebAsyncManager依赖RequestContextHolder中的请求上下文跟踪超时，切换到非Spring管理线程池时上下文丢失，WebAsyncManager无法维护自定义超时，Tomcat会触发默认30秒超时。
• 配置优先级逻辑：server.servlet.async.timeout优先级高于Tomcat的asyncTimeout，但上下文丢失时Spring无法更新超时状态，Tomcat会使用自身默认值。
• GraphQL上下文关联：GraphQL的asyncTimeout仅对顶层请求生效，若数据层异步操作脱离GraphQL上下文链，该配置无法覆盖底层超时，需确保整个响应链在GraphQL上下文内执行。

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