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是否在不断改进。在某些点之后，由于上下文切换而增加了更多的并发线程成本。

考虑复制那些什么的线程会做，可能使用模拟实际的API的您呼叫的时延远程存根。

另外，请考虑您可能会将下游系统推向其突破点。至少有一次，我们编写了非常并行的代码，只是为了发现我们遇到的下游外部API不是平行的，并且在我们的高峰期使用中导致了其他客户的中断。我们不小心DOS的API。

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