国产探花免费观看_亚洲丰满少妇自慰呻吟_97日韩有码在线_资源在线日韩欧美_一区二区精品毛片,辰东完美世界有声小说,欢乐颂第一季,yy玄幻小说排行榜完本

首頁 > 編程 > Java > 正文

支持生產阻塞的Java線程池

2019-11-26 15:32:34
字體:
來源:轉載
供稿:網友

通常來說,生產任務的速度要大于消費的速度。一個細節問題是,隊列長度,以及如何匹配生產和消費的速度。

一個典型的生產者-消費者模型如下:

 

在并發環境下利用J.U.C提供的Queue實現可以很方便地保證生產和消費過程中的線程安全。這里需要注意的是,Queue必須設置初始容量,防止生產者生產過快導致隊列長度暴漲,最終觸發OutOfMemory。

對于一般的生產快于消費的情況。當隊列已滿時,我們并不希望有任何任務被忽略或得不到執行,此時生產者可以等待片刻再提交任務,更好的做法是,把生產者阻塞在提交任務的方法上,待隊列未滿時繼續提交任務,這樣就沒有浪費的空轉時間了。阻塞這一點也很容易,BlockingQueue就是為此打造的,ArrayBlockingQueue和LinkedBlockingQueue在構造時都可以提供容量做限制,其中LinkedBlockingQueue是在實際操作隊列時在每次拿到鎖以后判斷容量。

更進一步,當隊列為空時,消費者拿不到任務,可以等一會兒再拿,更好的做法是,用BlockingQueue的take方法,阻塞等待,當有任務時便可以立即獲得執行,建議調用take的帶超時參數的重載方法,超時后線程退出。這樣當生產者事實上已經停止生產時,不至于讓消費者無限等待。

于是一個高效的支持阻塞的生產消費模型就實現了。

等一下,既然J.U.C已經幫我們實現了線程池,為什么還要采用這一套東西?直接用ExecutorService不是更方便?

我們來看一下ThreadPoolExecutor的基本結構:

 

可以看到,在ThreadPoolExecutor中,BlockingQueue和Consumer部分已經幫我們實現好了,并且直接采用線程池的實現還有很多優勢,例如線程數的動態調整等。

但問題在于,即便你在構造ThreadPoolExecutor時手動指定了一個BlockingQueue作為隊列實現,事實上當隊列滿時,execute方法并不會阻塞,原因在于ThreadPoolExecutor調用的是BlockingQueue非阻塞的offer方法:

復制代碼 代碼如下:

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
        if (runState == RUNNING && workQueue.offer(command)) {
            if (runState != RUNNING || poolSize == 0)
                ensureQueuedTaskHandled(command);
        }
        else if (!addIfUnderMaximumPoolSize(command))
            reject(command); // is shutdown or saturated
    }
}

這時候就需要做一些事情來達成一個結果:當生產者提交任務,而隊列已滿時,能夠讓生產者阻塞住,等待任務被消費。

關鍵在于,在并發環境下,隊列滿不能由生產者去判斷,不能調用ThreadPoolExecutor.getQueue().size()來判斷隊列是否滿。

線程池的實現中,當隊列滿時會調用構造時傳入的RejectedExecutionHandler去拒絕任務的處理。默認的實現是AbortPolicy,直接拋出一個RejectedExecutionException。

幾種拒絕策略在這里就不贅述了,這里和我們的需求比較接近的是CallerRunsPolicy,這種策略會在隊列滿時,讓提交任務的線程去執行任務,相當于讓生產者臨時去干了消費者干的活兒,這樣生產者雖然沒有被阻塞,但提交任務也會被暫停。

復制代碼 代碼如下:

public static class CallerRunsPolicy implements RejectedExecutionHandler {
    /**
     * Creates a <tt>CallerRunsPolicy</tt>.
     */
    public CallerRunsPolicy() { }

    /**
     * Executes task r in the caller's thread, unless the executor
     * has been shut down, in which case the task is discarded.
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            r.run();
        }
    }
}

但這種策略也有隱患,當生產者較少時,生產者消費任務的時間里,消費者可能已經把任務都消費完了,隊列處于空狀態,當生產者執行完任務后才能再繼續生產任務,這個過程中可能導致消費者線程的饑餓。

參考類似的思路,最簡單的做法,我們可以直接定義一個RejectedExecutionHandler,當隊列滿時改為調用BlockingQueue.put來實現生產者的阻塞:

復制代碼 代碼如下:

new RejectedExecutionHandler() {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                if (!executor.isShutdown()) {
                        try {
                                executor.getQueue().put(r);
                        } catch (InterruptedException e) {
                                // should not be interrupted
                        }
                }
        }
};

這樣,我們就無需再關心Queue和Consumer的邏輯,只要把精力集中在生產者和消費者線程的實現邏輯上,只管往線程池提交任務就行了。

相比最初的設計,這種方式的代碼量能減少不少,而且能避免并發環境的很多問題。當然,你也可以采用另外的手段,例如在提交時采用信號量做入口限制等,但是如果僅僅是要讓生產者阻塞,那就顯得復雜了。

發表評論 共有條評論
用戶名: 密碼:
驗證碼: 匿名發表
主站蜘蛛池模板: 涞水县| 南雄市| 桃园市| 同心县| 兴国县| 两当县| 金堂县| 苍山县| 荔浦县| 甘谷县| 左云县| 绍兴县| 滁州市| 永吉县| 阜新市| 皮山县| 阿图什市| 平果县| 龙陵县| 临城县| 华坪县| 涞水县| 清河县| 榆树市| 阳山县| 西宁市| 精河县| 顺义区| 崇仁县| 息烽县| 马山县| 瑞金市| 海盐县| 石景山区| 麻栗坡县| 屯昌县| 开江县| 哈尔滨市| 宜君县| 固安县| 承德市|