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是我們的瓶頸。