java - 为什么 Spring 使用 ForkPoolJoin 而不是带有 @Async 的 ThreadPoolTask​​Executor？

Question

为了学习，我正在研究 Spring Boot REST API。我应该在收到请求时减少代码的执行时间。所以，我认为让代码异步是个好主意。但是，不幸的是，我为此遇到了 Spring 的一些问题，尽管我花了几个小时在线研究以找到解决方案，但我什么也没找到。让我解释一下:

为了优化我的代码，我决定使用 @Async Spring Annotation。为此，我创建了一个如下所示的 AsyncConfiguration 类:

@Configuration
@EnableAsync
public class AsyncConfiguration {

    @Bean(name = "asyncExecutor")
    public Executor asyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(1);
        executor.setQueueCapacity(50);
        executor.setThreadNamePrefix("AsynchThread-");
        executor.initialize();
        return executor;
    }
    
}

通常，asyncExecutor() 方法返回 Spring 必须使用的 Executor 类或子类进行异步调用。在本例中，它是一个 ThreadPoolTask​​Executor。我的大部分代码都用 @Async 注释来使用我的 asyncExecutor，就像这样:

@Async("asyncExecutor")
    public CompletableFuture<User> calculateRewards(User user) {
        return CompletableFuture.supplyAsync(() -> {
            logger.info("Calculating Reward for user : " + user.getUserName());
            List<VisitedLocation> userLocations = user.getVisitedLocations();
            List<Attraction> attractions = gps.getAllAttraction();
        
            for(VisitedLocation visitedLocation : userLocations) {
                for(Attraction attraction : attractions) {
                    if(user.getUserRewards().stream().filter(r -> r.attraction.attractionName.equals(attraction.attractionName)).count() == 0) {
                        if(nearAttraction(visitedLocation, attraction)) {
                            user.addUserReward(new UserReward(visitedLocation, attraction, reward.getAttractionRewardPoints(attraction.attractionId, user.getUserId())));
                        }
                    }
                }
            }

            return user;
        });
    }

但是，重点是:当我运行代码时，Spring 不使用我的 asyncExecutor() bean。我怎么知道的？首先，当我调用带注释的方法时，在终端中，这是我看到的:

2022-10-27 00:26:24.688  INFO 41436 --- [onPool-worker-4] c.T.T.service.TourGuideMainService       : Calculating Reward for user : internalUser919

[onPool-worker-4]" 是线程名称，或者至少是线程名称的末尾。但它不应该这样命名。如果你看一下我上面的 asyncExecutor() 方法，你可以看到有一个 executor.setThreadNamePrefix("AsynchThread-");。如果代码按预期工作，线程应称为“AsynchThread-4”，但事实并非如此。

其次，我决定在 Debug 模式下运行我的代码，然后进入 VS Code 的 Debug 菜单，我发现了两件事:

1 - 当我运行同时调用 1000 个 calculateRewards() 的压力测试时，只创建了 11 个线程。考虑到 calculateRewards() 方法的执行时间以及执行器默认具有其 maxPoolSize 的事实(即 Integer.MAX_VALUE)，有应该超过 11 个线程；

2 - 线程创建时的完整名称是[ForkJoinPool.commonPool-worker-4]" ;

似乎 Spring 正在使用 ForkJoinPool 类来创建线程，它从来没有考虑过我的执行器的配置。我不知道它为什么这样做，我根本没有使用过 ForkJoinPool，就像我说的，我在网上搜索时没有找到任何东西。

那么，为什么 Spring 在我的代码中使用 ForkJoinPool 而不是 ThreadPoolTask​​Executor 作为异步方法？最重要的是:我该如何解决这个问题？

(我希望它是可以理解的...)

编辑 1:在一些随机测试中，我发现，如果 Spring 似乎使用一些 ForkJoinPool 线程来执行我的代码，它无论如何都会创建一个“AsynchThread”，但不要使用它。那就更糊涂了……-_-”

Answer 1

这里的问题是您正在使用 CompletableFuture.supplyAsync 来生成您的 CompletableFuture。

此方法将派生一个在 ForkJoinPool.commonPool() 中运行的任务，并在该任务中执行您的供应商。

由于 spring 将提供执行器并异步运行整个方法，因此您不需要将 supplyAsync 函数与 lambda 一起使用。相反，您的异步方法应该看起来像这样(在您的服务内部):

@Service
public class MyService {
  
  ...

  @Async("asyncExecutor")
  public CompletableFuture<User> calculateRewards(User user) {
    logger.info("Calculating Reward for user : " + user.getUserName());
    List<VisitedLocation> userLocations = user.getVisitedLocations();
    List<Attraction> attractions = gps.getAllAttraction();

    for(VisitedLocation visitedLocation : userLocations) {
      for(Attraction attraction : attractions) {
        if(user.getUserRewards().stream().filter(r -> r.attraction.attractionName.equals(attraction.attractionName)).count() == 0) {
          if(nearAttraction(visitedLocation, attraction)) {
            user.addUserReward(new UserReward(visitedLocation, attraction, reward.getAttractionRewardPoints(attraction.attractionId, user.getUserId())));
          }
        }
      }
    }

    return CompletableFuture.completedFuture(user);
  }

}

如果您将服务自动连接到执行您的方法的 CommandLineRunner 中，您可以使用此运行程序来查看 spring 将使用您拥有的 Executor 异步执行您的方法在您的配置中定义。

例如:

@Component
public class Runner implements CommandLineRunner {

  private final MyService service;

  public Runner(MyService service) {
    this.service = service;
  }

  @Override
  public void run(String... args) throws Exception {
    CompletableFuture<User> user1 = service.calculateRewards(new User("user1"));
    CompletableFuture<User> user2 = service.calculateRewards(new User("user2"));
    CompletableFuture<User> user3 = service.calculateRewards(new User("user3"));

    CompletableFuture.allOf(user1,user2,user3).join();
  }

}