ThreadPoolExecutor 的理解

公司有位大佬寫下這麼一段話

 

①，FixedThreadPool任務隊列的無邊界會致使內存溢出以及高延遲
②，CachedThreadPool線程數的無邊界會致使併發高的時候建立的線程數不可控

建議由於二者都不是特別友好，因此推薦直接使用ThreadPoolExecutor，它提供了不少參數能夠進行細粒度的控制，

而後我就找了個機會了解下，這是怎麼回事

查看ThreadPoolExecutor#execute 方法

簡介說，這可能啓動新的線程或原有線程開啓任務或拒絕任務，代碼以下

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();        
        int c = ctl.get();
//如何工做的線程數小於核心線程數
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))//新起一個線程並添加到工做隊列,而後return
                return;
            c = ctl.get();
        }
        //workQueue.offer將任務添加到工做隊列中
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        else if (!addWorker(command, false))
            reject(command);添加到工做隊列失敗，使用拒絕策略
    }

1，當執行線程數小於核心線程數或阻塞隊列滿的時候，會嘗試直接開啓一線程運行任務：addWorker

new Worker 會從線程池中獲取新的線程，並保存到一個實例變量中，以下

Worker(Runnable firstTask) {
            setState(-1); // inhibit interrupts until runWorker
//記錄好第一個任務
            this.firstTask = firstTask;
//這裏實現線程的重用
            this.thread = getThreadFactory().newThread(this);
        }

線程運行時流程worker.t.start->worker#run->ThreadPoolExecutor#runWorker

會一直從workQueue取出任務，當workQueue爲空時，則阻塞獲取

2，當執行線程數大於等於核心線程數時，將任務添加到阻塞隊列：workQueue.offer(command)

 work工做時會運行runWorker方法，以下

final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
            //不斷從工做隊列中獲取task，並執行，這就是所謂的線程池和工做竊取機制
            while (task != null || (task = getTask()) != null) {
                w.lock();               
                if ((runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                    wt.interrupt();
                try {
                    beforeExecute(wt, task);
                    Throwable thrown = null;
                    try {
                        task.run();
                    } catch (RuntimeException x) {
                        thrown = x; throw x;
                    } catch (Error x) {
                        thrown = x; throw x;
                    } catch (Throwable x) {
                        thrown = x; throw new Error(x);
                    } finally {
                        afterExecute(task, thrown);
                    }
                } finally {
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }