常用线程池
ThreadPoolExecutor解析
- 源码分析
- 流程图：
- 线程池处理流程
- execute方法：
- addWorker方法
- runwoker方法：
- getTask方法：
- processWorkerExit方法：
- ThreadPoolExecutor内部有实现4个拒绝策略：
ScheduledThreadPoolExecutor

常用线程池

ThreadPoolExecutor解析

阿里不推荐使用：
内部的无界队列，造成oom
自定义线程池：
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(10, 20, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(10));

源码分析

流程图：

线程池处理流程

private static final int COUNT_BITS = Integer.SIZE - 3;
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
// runState is stored in the high-order bits
    private static final int RUNNING    = -1 << COUNT_BITS;
    private static final int SHUTDOWN   =  0 << COUNT_BITS;
    private static final int STOP       =  1 << COUNT_BITS;
    private static final int TIDYING    =  2 << COUNT_BITS;
    private static final int TERMINATED =  3 << COUNT_BITS;

COUNT_BITS：int位数 = 32-3
实际上使用32位中的高3位表示

execute方法：

int c = ctl.get();
1、当前的线程数是否小于corePoolSize，如果是，
使用入参任务通过addWord方法创建一个新的线程，
如果能完成新线程创建exexute方法结束，成功提交任务；
if (workerCountOf(c) < corePoolSize) {
    if (addWorker(command, true))
        return;
    c = ctl.get();
}
2、没有完成任务提交；状态为运行并且能成功加入任务到工作队列后，
再进行一次check，如果状态在任务加入队列后变为了非运行（有可能是在执行到这里线程池shutdown了）
非运行状态下当然是需要reject；
然后再判断当前线程数是否为0（有可能这个时候线程数变为了0），如是，新增一个线程；
if (isRunning(c) && workQueue.offer(command)) {
    int recheck = ctl.get();
    if (! isRunning(recheck) && remove(command))
        reject(command);
    else if (workerCountOf(recheck) == 0)//判断当前工作线程池数是否为0  
        addWorker(null, false); 
        //如果是创建一个null任务，任务在堵塞队列存在了就会从队列中取出 这样做的意义是
        //保证线程池在running状态必须有一个任务在执行
}
3、如果不能加入任务到工作队列，将尝试使用任务新增一个线程，如果失败，
则是线程池已经shutdown或者线程池已经达到饱和状态，所以reject;
拒绝策略不仅仅是在饱和状态下使用，
在线程池进入到关闭阶段同样需要使用到；：
else if (!addWorker(command, false))
    reject(command);

z总结：

如果当前线程数小于corePoolSize, 通过addWorker创建新线程，
如果能完成新线程创建execute方法结束，成成功提交任务
如果上一步没有完成提交任务，状态为运行态并且假如任务队列，再check一次
如果状态再任务假如队列后变为非运行（线程池挂了），就执行reject,
判断当前线程数是否为0，如果没了，新增一个线程
如果不能加入道工作队列，就尝试使用任务新增线程，如果线程池已经shutdown或满了，
也会reject,
reject：饱和状态下，线程池关闭阶段也会用到

addWorker方法

private boolean addWorker(Runnable firstTask, boolean core) {
    retry: goto写法 用于重试
    for (;;) { 
        int c = ctl.get();
        int rs = runStateOf(c);
        // Check if queue empty only if necessary.
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
               线程状态非运行并且非shutdown状态任务为空，队列非空就不能新增线程了
            return false;
        for (;;) {
            int wc = workerCountOf(c);
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                当前现场大于等于最大值 
                等于核心线程数 非核心大于等于线程池数 说明达到了阈值 
                最大线程数 就不新增线程
                return false;
            if (compareAndIncrementWorkerCount(c)) ctl+1 工作线程池数量+1 如果成功
            就跳出死循环。
            cas操作 如果为true 新增成功 退出
                break retry;
            c = ctl.get();  // Re-read ctl
            if (runStateOf(c) != rs)
                continue retry; 进来的状态和此时的状态发生改变 重头开始 重试 
            // else CAS failed due to workerCount change; retry inner loop
        }
    }
    上面主要是对ctl工作现场+1
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        w = new Worker(firstTask); 内部类 封装了线程和任务 通过threadfactory创建线程
        final Thread t = w.thread; 毎一个worker就是一个线程数
        if (t != null) {
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // Recheck while holding lock.
                // Back out on ThreadFactory failure or if
                // shut down before lock acquired.
                重新获取当前线程状态
                int rs = runStateOf(ctl.get());
                 小于shutdown就是running状态
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                        SHUTDOWN 和firstTask 为空是从队列中处理任务 那就可以放到集合中
                       线程还没start 就是alive就直接异常
                     if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    workers.add(w);
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s; 记录最大线程数
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                t.start(); 启动线程
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);//失败回退 从wokers移除w 线程数减1 尝试结束线程池
    }
    return workerStarted;
}

private final class Worker
    extends AbstractQueuedSynchronizer
    implements Runnable
{
    /**
     * This class will never be serialized, but we provide a
     * serialVersionUID to suppress a javac warning.
     */
    private static final long serialVersionUID = 6138294804551838833L;
    /** Thread this worker is running in.  Null if factory fails. */
    正在运行woker线程
    final Thread thread;
    /** Initial task to run.  Possibly null. */
    传入的任务
    Runnable firstTask;
    /** Per-thread task counter */
    完成的任务数 监控用
    volatile long completedTasks;
    /**
     * Creates with given first task and thread from ThreadFactory.
     * @param firstTask the first task (null if none)
     */
    Worker(Runnable firstTask) {
        禁止线程中断 
        setState(-1); // inhibit interrupts until runWorker
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }
    /** Delegates main run loop to outer runWorker  */
    public void run() {
        runWorker(this);
    }

runwoker方法：

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();//获取当前线程
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts 把state从-1改为0 意思是可以允许中断
    boolean completedAbruptly = true;
    try { task不为空 或者阻塞队列中拿到了任务
        while (task != null || (task = getTask()) != null) {
            w.lock();
            // If pool is stopping, ensure thread is interrupted;
            // if not, ensure thread is not interrupted.  This
            // requires a recheck in second case to deal with
            // shutdownNow race while clearing interrupt
            如果当前线程池状态等于stop 就中断
            //Thread.interrupted() 中断标志
            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++; 完成任务数+1
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally { 
        processWorkerExit(w, completedAbruptly);
    }
}

getTask方法：

private Runnable getTask() {
    boolean timedOut = false; // Did the last poll() time out?
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);//获取线程池运行状态
            shuitdown或者weikong 那就工作现场-1 同事返回为null 
        // Check if queue empty only if necessary.
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }
             重新获取工作线程数
        int wc = workerCountOf(c);
        timed是标志超时销毁
        allowCoreThreadTimeOut  true 核心线程池也是可以销毁的
        // Are workers subject to culling?
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }
        try {
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

processWorkerExit方法：

private void processWorkerExit(Worker w, boolean completedAbruptly) {
    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
        decrementWorkerCount();
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        completedTaskCount += w.completedTasks;
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
    int c = ctl.get();
    if (runStateLessThan(c, STOP)) {
        if (!completedAbruptly) {
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        addWorker(null, false);
    }
}

ThreadPoolExecutor内部有实现4个拒绝策略：

(1)、CallerRunsPolicy，由调用execute方法提交任务的线程来执行这个任务；
(2)、AbortPolicy，抛出异常RejectedExecutionException拒绝提交任务；
(3)、DiscardPolicy，直接抛弃任务，不做任何处理；
(4)、DiscardOldestPolicy，去除任务队列中的第一个任务（最旧的），重新提交；
实际上需要自定义处理

public class MyRejectedExecutionHandler implements RejectedExecutionHandler {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
    }
}

ScheduledThreadPoolExecutor