如何讓生產者線程等待消費者線程完成

Question

每個操作都將在我們的應用程序中創建主線程，並且此線程將創建任務並由提交ExecutorService。

當前這些生產者線程會將任務添加到ArrayList，從threadPool的線程將執行任務。

在高負載（當有很多操作時），我得到「JVM內存不足」錯誤。

我怎麼能通過讓主線等待一段時間基於一些限制來避免這種情況？

Answer 1

內存不足是另一個問題，但你想要producer/consumer with blocking queue。

如果在執行程序池中消耗線程無法足夠快地處理請求，則阻塞隊列將被填滿。一旦它滿了，生產者將被阻止添加新的任務，直到消費者可以處理請求並騰出空間。

這可能無法解決您的OOM問題。您需要剖析代碼以查看內存正在被消耗的位置。也許你的消費者線程在完成時沒有清理，或者GC沒有收回靜態數據。

這可能是這樣的情況，如Sorcerer's Apprentice。如果生產者如此迅速地創建請求而不停止，直到所有內存耗盡，您都會收到OOM錯誤。如果這是根本原因，那麼阻塞deque將修復它。

Answer 2

如果您正在使用ExecutorService執行任務，那麼我會建議您，而不是默認實現的線程池，您可以通過實例化具有所需配置的ThreadPoolExecutor來定義您自己的具有自定義配置的線程池。請看下圖：

/** 
    * Creates a new {@code ThreadPoolExecutor} with the given initial 
    * parameters and default thread factory and rejected execution handler. 
    * It may be more convenient to use one of the {@link Executors} factory 
    * methods instead of this general purpose constructor. 
    * 
    * @param corePoolSize the number of threads to keep in the pool, even 
    *  if they are idle, unless {@code allowCoreThreadTimeOut} is set 
    * @param maximumPoolSize the maximum number of threads to allow in the 
    *  pool 
    * @param keepAliveTime when the number of threads is greater than 
    *  the core, this is the maximum time that excess idle threads 
    *  will wait for new tasks before terminating. 
    * @param unit the time unit for the {@code keepAliveTime} argument 
    * @param workQueue the queue to use for holding tasks before they are 
    *  executed. This queue will hold only the {@code Runnable} 
    *  tasks submitted by the {@code execute} method. 
    * @throws IllegalArgumentException if one of the following holds:<br> 
    *   {@code corePoolSize < 0}<br> 
    *   {@code keepAliveTime < 0}<br> 
    *   {@code maximumPoolSize <= 0}<br> 
    *   {@code maximumPoolSize < corePoolSize} 
    * @throws NullPointerException if {@code workQueue} is null 
    */ 
    public ThreadPoolExecutor(int corePoolSize, 
           int maximumPoolSize, 
           long keepAliveTime, 
           TimeUnit unit, 
           BlockingQueue<Runnable> workQueue) { 
    } 

在這裏，你可以提供自己的BlockingQueue由您定義的大小，所以現在在這種情況下，當隊列已滿再往新任務會等到隊列中有空間接受新任務。

通過此實施，您可以按照您的要求控制和調整線程池。

您可以點擊此link更詳細的信息，你可能會覺得很無聊：P，但請相信我，你會發現它非常有用，一旦你讀它，只是閱讀耐心.. :)