Spring Boot for循环内调用Async异步方法的最优实现方案咨询

实体模型 #

package com.example.asyncmethod;

import com.fasterxml.jackson.annotation.JsonIgnoreProperties;

@JsonIgnoreProperties(ignoreUnknown=true)
public class User {

  private String name;
  private String blog;

  public String getName() {
    return name;
  }

  public void setName(String name) {
    this.name = name;
  }

  public String getBlog() {
    return blog;
  }

  public void setBlog(String blog) {
    this.blog = blog;
  }

  @Override
  public String toString() {
    return "User [name=" + name + ", blog=" + blog + "]";
  }

}

原有异步服务类 #

package com.example.asyncmethod;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.web.client.RestTemplateBuilder;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

import java.util.concurrent.CompletableFuture;

@Service
public class GitHubLookupService {

  private static final Logger logger = LoggerFactory.getLogger(GitHubLookupService.class);

  private final RestTemplate restTemplate;

  public GitHubLookupService(RestTemplateBuilder restTemplateBuilder) {
    this.restTemplate = restTemplateBuilder.build();
  }

  @Async
  public CompletableFuture<User> findUser(Integer user) throws InterruptedException {
    logger.info("Looking up " + user);
    String url = String.format("https://api.github.com/users/%s", user);
    User results = restTemplate.getForObject(url, User.class);
    
    Thread.sleep(1000L);
    return CompletableFuture.completedFuture(results);
  }

}

原有执行启动类 #

package com.example.asyncmethod;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.util.concurrent.CompletableFuture;

@Component
public class AppRunner implements CommandLineRunner {

  private static final Logger logger = LoggerFactory.getLogger(AppRunner.class);

  private final GitHubLookupService gitHubLookupService;

  public AppRunner(GitHubLookupService gitHubLookupService) {
    this.gitHubLookupService = gitHubLookupService;
  }

  @Override
  public void run(String... args) throws Exception {
    // Start the clock
    long start = System.currentTimeMillis();

    // in real use case before the for loop I will have a database call to get the number of records and based on the size I need to call that many number of times.
    for(int i=0; i <= 10; i++){
      CompletableFuture<User> page1 = gitHubLookupService.findUser(1);
    }

    // Wait until they are all done
    CompletableFuture.allOf(page1).join();

    // Print results, including elapsed time
    logger.info("Elapsed time: " + (System.currentTimeMillis() - start));
    logger.info("--> " + page1.get());

  }

}

实际业务会先查询数据库获取记录数，按返回数量调用对应次数的异步方法，后续还会新增findOrders、findInventory、findAccess等多个跨HTTP服务调用的异步方法，需要解决部分请求成功部分失败的异步调用可用性问题。

1. 修复异步任务收集逻辑，添加单任务异常兜底 #

原有代码的异步任务是for循环内的局部变量，外部无法聚合，且单个任务失败会直接终止所有任务等待逻辑。优化后将所有异步任务存入集合，每个任务单独配置异常兜底，避免单个失败影响整体流程：

@Override
public void run(String... args) throws Exception {
    long start = System.currentTimeMillis();
    // 模拟查库得到的需要查询的用户ID列表
    List<Integer> userIdList = databaseMapper.queryUserIds();
    List<CompletableFuture<User>> futures = new ArrayList<>();

    for (Integer userId : userIdList) {
        CompletableFuture<User> future = gitHubLookupService.findUser(userId)
                // 单任务异常捕获，不影响其他任务执行
                .exceptionally(ex -> {
                    logger.error("用户{}查询失败: {}", userId, ex.getMessage(), ex);
                    // 可根据业务需求返回兜底对象，或封装带错误标识的结果类
                    User fallbackUser = new User();
                    fallbackUser.setName("查询失败_" + userId);
                    return fallbackUser;
                });
        futures.add(future);
    }

    // 等待所有任务执行完成（成功/失败都算完成）
    CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

    // 收集所有结果
    List<User> resultList = futures.stream()
            .map(CompletableFuture::join)
            .toList();

    logger.info("Elapsed time: " + (System.currentTimeMillis() - start));
    // 可自行统计成功/失败数量，做后续业务处理
}

2. 自定义IO密集型异步线程池 #

Spring默认的@Async使用无复用的线程池，高并发场景下会导致OOM，针对跨HTTP调用的IO密集型场景，单独配置专用线程池统一管理资源：

@Configuration
@EnableAsync
public class AsyncThreadPoolConfig {
    @Bean("httpInvokeAsyncExecutor")
    public ThreadPoolTaskExecutor httpAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        // IO密集型场景核心线程数可设置为CPU核心数*2以上，可根据压测结果调整
        executor.setCorePoolSize(20);
        executor.setMaxPoolSize(60);
        executor.setQueueCapacity(300);
        executor.setThreadNamePrefix("http-async-");
        // 拒绝策略设置为调用方线程执行，避免任务丢失，也可根据业务选择丢弃策略
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }
}

在异步方法上指定使用该线程池，后续新增的其他HTTP异步调用方法也可复用该配置：

@Async("httpInvokeAsyncExecutor")
public CompletableFuture<User> findUser(Integer user) throws InterruptedException {
    // 原有逻辑不变
}

3. 配置HTTP调用超时，避免线程阻塞 #

为RestTemplate添加超时配置，避免远程服务响应异常导致线程长时间挂起，占满线程池资源：

public GitHubLookupService(RestTemplateBuilder restTemplateBuilder) {
    this.restTemplate = restTemplateBuilder
            .setConnectTimeout(Duration.ofSeconds(3)) // 连接超时3秒
            .setReadTimeout(Duration.ofSeconds(5)) // 读取超时5秒
            .build();
}

4. 通用能力扩展 #

后续新增其他异步调用方法时，可封装通用的异步调用工具类，统一处理异常捕获、日志打印、指标统计逻辑，避免重复编码。任务全部执行完成后可根据失败率配置告警、降级逻辑，比如失败率超过30%时返回缓存兜底数据。

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