java - Spring:使用 ThreadPoolTask​​Executor 创建真正可扩展的线程池

Question

我有一个使用 Spring ThreadPoolTask​​Executor 的 REST API。 API 在内部将可调用作业提交给 ThreadPoolTask​​Executor，后者调用第三方。调用完成后，将执行业务逻辑并将结果返回给调用者。

代码工作正常，但当负载增加时，性能变得非常糟糕。我怀疑这是 ThreadPoolTask​​Executor 线程池大小的结果。因此，如果并发用户数为 n，但我们只有 x 个线程(x 小于 n)，那么 x 线程将不得不不必要地等待有限数量的线程来处理其请求。

我想并行处理第三方调用，但不想创建具有大量线程的线程池。

我的选择是使用 Executors.newFixedThreadPool(y)。在方法内部使用它，一旦过程完成就关闭对象。这是可能的，但不确定它的副作用，例如从方法创建固定线程池是否是一个好的做法。

其他选项可能是使用某种对象池(例如 GenericObjectPoolConfig)并使用它来获取线程。

其他选项可能是将最大池大小设置为 Integer.max 并将队列容量减少到 1。这样，每次新请求到来时，它都会创建一个新线程，而不是将对象存储在队列中。

ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
    threadPoolTaskExecutor.setCorePoolSize(20);
    threadPoolTaskExecutor.setMaxPoolSize(Integer.MAX_VALUE);
    threadPoolTaskExecutor.setWaitForTasksToCompleteOnShutdown(true);
    threadPoolTaskExecutor.setQueueCapacity(1);
    threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
    threadPoolTaskExecutor.initialize();

如果有人能分享他的想法，将会很有帮助。

@Configuration
public class TestConfiguration{

   @Bean
public ConcurrentTaskExecutor concurrentTaskExecutor() {
    ConcurrentTaskExecutor concurrentTaskExecutor = new ConcurrentTaskExecutor();
    concurrentTaskExecutor.setConcurrentExecutor(getExecutor());
    return concurrentTaskExecutor;
}

private Executor getExecutor() {
    ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
    threadPoolTaskExecutor.setCorePoolSize(20);
    threadPoolTaskExecutor.setMaxPoolSize(30);
    threadPoolTaskExecutor.setWaitForTasksToCompleteOnShutdown(true);
    threadPoolTaskExecutor.setQueueCapacity(75);
    threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
    threadPoolTaskExecutor.initialize();
    return threadPoolTaskExecutor;
}
}

@Service
public class TestServiceImpl{
  @Autowired
  private ConcurrentTaskExecutor concurrentTaskExecutor;

  @Override
  @Transactional
  public DTO getDTO() {
    Callable<TESTDTO> test1Callable = new Test1Callable();
    Future<TESTDTO> testDTO1 = concurrentTaskExecutor.submit(test1Callable);

    Callable<TESTDTO> test2Callable = new Test2Callable();
    Future<TESTDTO> testDTO2 =concurrentTaskExecutor.submit(test2Callable);

    Callable<TESTDTO> test3Callable = new Test3Callable();
    Future<TESTDTO> testDTO3 =concurrentTaskExecutor.submit(test3Callable);

    // Perform logic on DTO's
    return DTO;
  }

Answer 1

我真的觉得你的东西看起来不错。您需要做的就是进行良好的性能测试来尝试队列容量和池大小的各种值。就此而言，任何其他领域。

话虽如此，您可以遵循一些一般准则。例如，如果线程正在进行 CPU 密集型计算，那么您可能希望将池大小限制为 2* 核心数或该附近的某个值。如果线程正在处理 IO 密集型任务(例如与外部 API 通信)，那么您可以走得更高。在我自己过去的测试中，我注意到，在英特尔双核上执行 IO 密集型任务时，投资返回率约为 50 个线程。 50 岁之后，我将不再看到任何性能提升。但请自行测试并验证，因为 CPU 一直在改进。在某些时候，由于上下文切换，添加更多并发线程的成本过高。

考虑重复这些线程将要执行的操作，可能使用远程 stub 来模拟您正在调用的实际 API 的延迟。

此外，请考虑您可能会将下游系统推向极限。至少有一次，我们编写了非常并行的代码，却发现我们所使用的下游外部 API 并不那么并行，并且在我们的高峰使用期间导致了其他客户的中断。我们不小心对 API 进行了 DOS 操作。

请分析您的系统并检查 CPU、内存或磁盘 IO 是否紧张。如果内存、CPU 和磁盘 IO 都正常，那么可能会遇到下游限制。有一次，我们在一些高度并行的代码中记录了太多日志，而磁盘 IO 是我们的瓶颈。