首先，在Java中，创建线程有如下四种方式

继承Thread
实现Runnable
实现Callable
使用线程池

线程池

为什么需要线程池
线程池的工作主要是控制运行的线程数量，处理过程中将任务放入队列，然后在线程创建后启动这些任务，如果线程的数量超出了线程池的数量，超出的线程排队等待，等待其他线程执行完毕，再从队列中取出任务进行执行。

线程池的特点：线程服用，控制最大的并发数，管理线程
降低资源消耗，通过重复利用已创建的线程降低线程的创建和销毁造成的损耗。
提高响应速度，当任务到达时，任务可以不需要等待线程创建而直接从线程池中取出立即执行。
提供线程的可管理性，线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行同一的分配，调优和监控。
何时使用线程池

单个任务处理时间比较短
需要处理的任务数量很大
Java中线程池的实现
通过Executor框架实现的，该框架用到了Executor，Executors，ExecutorService，ThreadPoolExecutor这几个类。
Java中线程池的继承图

Java中线程池的分类
总共有5种，底层通过ThreadPoolExecutor以及BlockingQueue来实现的

延时类线程池(ScheduledExecutorService) Executors.newSingleThreadScheduledExecutor()
使用当前机器上可用的处理器作为他的并行级别(ForkJoinPool) Executors.newWorkStealingPool(int parallelism)
创建定长的线程池 Executors.newFixedThreadPool()
创建单个线程的线程池 Executors.newSingleThreadExecutor()
创建可扩容的线程池 Executos.newCachedThreadPool()

通过上面的五种线程池的源码发现，都是基于ThreadPoolExecutor这个类进行实现的

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

ThreadPoolExecutor的原理

从构造函数说起

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.acc = System.getSecurityManager() == null ?
                null :
                AccessController.getContext();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

其中 ThreadPoolExecutor 中有7大参数，分别为如下几个

corePoolSize 工作的核心线程数(常驻线程)
maximumPoolSize 最大的线程数(当核心线程数不足时，将会扩充线程，达到最大的线程数)
keepAliveTime 大于corePoolSize线程的空闲时间，当当前线程池的线程数大于核心线程数，且空闲时间大于keepAliveTime时，将会回收大于 corePoolSize的线程
unit 大于corePoolSize线程的空闲时间单位
workQueue 任务队列，被提交但尚未被执行的任务。（候客区）
threadFactory 表示生成线程池中工作线程的线程工厂，用于创建线程一般用默认的即可。
handler 拒绝策略，表示当队列满了并且工作线程大于等于线程池的最大线程数。（当办理窗口已满，候客区已满，将会采用此拒绝策略）
线程池的工作原理

1. 在创建了线程池之后，等待提交过来任务请求。
2. 当调用execute()方法添加一个请求任务时，线程池会做如下判断：
2.1 如果正在运行的线程数量小于corePoolSize，那么立马创建线程执行这个任务
2.2 如果正在运行的线程数量大于或等于corePoolSize，那么将这个任务放入阻塞队列中等待执行。
2.3 如果这时候队列已满，且正在运行的线程数量小于maximumPoolSize，那么创建非核心线程立刻运行这个任务。
2.4 如果队列已满，且正在运行的线程数大于或等于maximumPoolSize，那么线程池将会启动饱和拒绝策略来执行。
3. 当一个线程完成任务后，会从队列中取出下一个任务来执行。
4. 当一个线程无事可做超过一定时间（keepAliveTime）时，线程池将会判断：
4.1 如果当前运行的线程数大于corePoolSize，那么这个线程将会被停止。
4.2 线程池的所有任务完成之后，他的大小将会收缩到corePoolSize的大小
线程池的拒绝策略
阻塞队列已满，且线程池的线程池的线程大小已经达到了maximumPoolSize,这时需要拒绝策略机制合理的处理这个问题
AbortPolicy（默认）：直接跑出RejectedExecutionException异常阻止系统正常运行。
CallerRunsPolicy：“调用者运行”一种调节机制，该策略既不会抛弃任务，也不会抛出异常，而是将某些任务回退到调用者，从而降低新任务的流量。
DiscardOldestPolicy：抛弃队列中等待最久的任务，然后把当前任务加入队列中尝试再次提交当前任务。
DiscardPolicy：直接丢弃任务，不予任何处理也不抛出异常。如果允许任务丢失，这是最好的一种方案。
以上内置拒绝策略均实现了RejectedExecutionHandler接口
线程池的选型
不使用内置的Executors提供的线程池，通过 ThreadPoolExecutor 自己创建

线程池的使用
采用 AbortPolicy 拒绝策略
```java public static void main(String[] args) { ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
```
 2,
 5,
 1L,
 TimeUnit.SECONDS,
 new LinkedBlockingDeque<>(3),
 Executors.defaultThreadFactory(),
 new ThreadPoolExecutor.AbortPolicy()
```
); try {
```
 for (int i = 0; i < 20; i++) {
     threadPoolExecutor.execute(() -> System.out.println(Thread.currentThread().getName() + "\t do"));
 }
```
}finally {
```
 threadPoolExecutor.shutdown();
```
} }

pool-1-thread-1 do pool-1-thread-3 do pool-1-thread-2 do pool-1-thread-4 do pool-1-thread-1 do pool-1-thread-4 do pool-1-thread-2 do pool-1-thread-3 do pool-1-thread-5 do Exception in thread “main” java.util.concurrent.RejectedExecutionException: Task cn.spectrumrpc.juc.ThreadPoolTests$$Lambda$1/1831932724@7699a589 rejected from java.util.concurrent.ThreadPoolExecutor@58372a00[Running, pool size = 5, active threads = 1, queued tasks = 0, completed tasks = 8] at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063) at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830) at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1379) at cn.spectrumrpc.juc.ThreadPoolTests.main(ThreadPoolTests.java:27)

当for循环的数小于maximumPoolSize+ blockingQueue的大小时，不会程序正常运行，且当数量大于corePoolSize的时候，会进行扩容，当大于maximumPoolSize+ blockingQueue的时候，如果采取AbortPolicy策略，将会抛出RejectedExecutionException
<a name="RLUOw"></a>
### 如果采取CallerRunsPolicy策略，将会出现main线程的情况，即返回给调用者
```java
pool-1-thread-1     do
pool-1-thread-2     do
pool-1-thread-2     do
pool-1-thread-1     do
pool-1-thread-2     do
pool-1-thread-1     do
pool-1-thread-2     do
pool-1-thread-1     do
pool-1-thread-2     do
pool-1-thread-3     do
pool-1-thread-3     do
pool-1-thread-4     do
pool-1-thread-5     do
pool-1-thread-2     do
pool-1-thread-1     do
main     do
pool-1-thread-3     do
pool-1-thread-4     do
pool-1-thread-3     do
pool-1-thread-5     do

如果采取DiscardOldestPolicy或DiscardPolicy策略

将会丢弃某些任务

如何合理配置线程池，你是如何考虑的：

1. CPU密集型

CPU密集的意思是该任务需要大量的运算，而没有阻塞，CPU一直运行。
CPU密集任务只有在真正的多核CPU上才可能得到加速（通过多线程）
CPU密集型任务配置尽可能少的线程数量：
公式： CPU核数 + 1 个线程的线程池

2. IO密集型

2.1 由于IO密集型任务线程并不是一直在执行任务，则应配置尽可能多的线程，如CPU核数*2
2.2 IO密集型，即该任务需要大量的IO，即会阻塞，在单线程上运行IO密集型的任务会导致浪费大量的CPU,运算能力浪费在等待。所以在IO密集型任务中使用多线程可以大大的加速程序运行，即使在单核CPU上，这种加速主要就是利用了被浪费的阻塞时间。
IO密集型，大部分线程都在阻塞，所以需要多配置线程数
参考公式： CPU核数/(1 - 阻塞系数) 阻塞系数在0.8-0.9之间
如： 8核CPU 8/(1 -0.9) = 80个线程

ThreadPoolExecutor 源码分析

线程池的重要属性及方法

// ctl变量，是线程池中的一个重要属性，将一个int变量分为了两部分 高3位标识当前线程池的state，
// 低29位标识当前的最大线程池(所以最大的线程数为2^29-1)
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// ctl变量中表示线程数的比特位(29)
private static final int COUNT_BITS = Integer.SIZE - 3;
// 线程池的最大线程容量(1<< 29) - 1，这就是线程池的最大容量
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
// 以下是线程池的状态
// running，运行状态  111(29个0)
private static final int RUNNING    = -1 << COUNT_BITS; 
// shutdown，关闭状态  000(29个0)
private static final int SHUTDOWN   =  0 << COUNT_BITS;
// stop，停止状态  001(29个0)
private static final int STOP       =  1 << COUNT_BITS;
// TIDYING，终止回收状态  010(29个0)
private static final int TIDYING    =  2 << COUNT_BITS;
// TERMINATED，彻底终止状态 011(29个0)
private static final int TERMINATED =  3 << COUNT_BITS;

计算runState以及当前的工作线程数方法

// 拿 ctl变量 与 容量 (1<<29) -1进行运算，求出低29位
private static int runStateOf(int c)     { return c & ~CAPACITY; }
// 拿 ctl变量 与  容量 (1<<29) -1进行运算，求出高3位
private static int workerCountOf(int c)  { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }

execute方法

public void execute(Runnable command) {
    // 如果传入的runnable为null，抛出异常
    if (command == null)
        throw new NullPointerException();
    // 获取ctl 变量的值
    int c = ctl.get();
    // 通过workerCountof(c)求出当前的运行线程数，判断是否小于当前的核心线程数
    if (workerCountOf(c) < corePoolSize) {
        // 如果小于当前的核心线程数，通过addWroker继续创建线程数，执行这个任务
        if (addWorker(command, true))
            return;
        // 如果添加任务失败，重新获取ctl的值
        c = ctl.get();
    }
    // 判断当前线程池是否处于running状态，且放入工作队列成功
    if (isRunning(c) && workQueue.offer(command)) {
        // 再次获取ctl的值
        int recheck = ctl.get();
        // 如果当前的线程池状态不是running，那就将之前的任务移除，采用拒绝策略进行拒绝
        if (! isRunning(recheck) && remove(command))
            reject(command);
        // 否则，当前线程池的状态良好，获取当前可用的线程数，如果为0，创建新的线程
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 如果线程池的状态不是running，或者移除队列失败了的话
    else if (!addWorker(command, false))
        reject(command);
}

addWoker方法

addWorker方法的两个参数
firstTask：将要执行的任务(如果传递为null，在runWorker就直接跳出循环了，仅代表创建一个线程，但是什么都不干)
core：是否为核心线程(如果传递为true，workerCount与corePoolSize比较，传递为false，workerCount与maximumPoolSize相比)

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        // 获取ctl变量，并获取当前的runState
        int c = ctl.get();
        int rs = runStateOf(c);
        // 如果runState >= SHUTDOWN,在线程池的状态中只有RUNNING不符合这个条件，
        // 或者为SHUTDOWN状态，因为shutdown还能处理已添加的队列，
        // 所以，firstTask==null，且workQueue不空的话，就能继续处理
        // firstTask==null，代表仅创建线程
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;
        for (;;) {
            int wc = workerCountOf(c);
            // 计算出当前的线程数，如果大于最大容量，返回false
            // 或者根据传入的core，判断是核心线程 or 最大线程，如果不符合返回false
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // 如果当前的工作线程符合条件，还没有达到阈值的话，那就cas，增加workerCount的数量
            // 然后跳出循环，执行封装worker的步骤(带标签的break，直接跳到了retry外面去)
            if (compareAndIncrementWorkerCount(c))
                break retry;
            // cas失败，重新跑一遍，增加workerCount
            c = ctl.get();  // Re-read ctl
            if (runStateOf(c) != rs)
                continue retry;
        }
    }
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        // 将firstTask封装为 Worker对象
        w = new Worker(firstTask);
        final Thread t = w.thread;
        // 如果 
        if (t != null) {
            // 获取锁，加锁
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                int rs = runStateOf(ctl.get());
                // 判断 runState 是否是running(rs < SHUTDOWN)
                // 或者 是SHUTDOWN且传入的task为null(SHUTDOWN不会处理新增任务，但是workQueue会处理完)
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    // 将其保存到 workers集合中
                    workers.add(w);
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                // 最终unlock
                mainLock.unlock();
            }
            // 如果添加到workers中成功，则 t.start 开始执行
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        // 
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

t.start()

t 为 Worker对象中的属性，在构造worker的是否，通过 getThreadFactory().newThread(this)来创建，因为 worker本身继承了 Runnable 所以，执行时调用的run方法，为 Worker中的run方法

public void run() {
    runWorker(this);
}

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 {
        // 如果firstTask 为null，则从workerQueu中取出一个进行处理
        // 否则，就先处理firstTask中的任务
        // 循环从workerQueue中取出任务
        while (task != null || (task = getTask()) != null) {
            // 加锁
            w.lock();
            // 如果runState 是stop，TIDYING，TERMINATED 三种状态的话
            // 这三种状态不会处理任何的任务了，所以，直接中断当前线程(wt.interrupt())
            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方法
                    // 如果发生了响应的异常，往外面抛
                    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();
            }
        }
        // 如果workerQueu中没有任务了，则设置completedAbruptly = false
        completedAbruptly = false;
    } finally {
        最终执行完成之后，执行 processWorkerExit，将执行结果 completedAbruptly传入
        processWorkerExit(w, completedAbruptly);
    }
}

processWorkerExit()

private void processWorkerExit(Worker w, boolean completedAbruptly) {
    // 如果异常执行完，completedAbruptly=true
    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
        // cas，减少当前的线程数
        decrementWorkerCount();
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        // 增加执行完成的任务数
        completedTaskCount += w.completedTasks;
        // 将worker从集合中删除
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
    int c = ctl.get();
    // runState 小于STOP，说明是 shutdown或者running状态
    if (runStateLessThan(c, STOP)) {
        // 如果正常执行完， completedAbruptly = false，条件成立，
        if (!completedAbruptly) {
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            // 默认，拿到的是 corePoolSize，这个条件不成立
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            // 如果当前的workerCount > corePoolSize，就不需要再次添加worker进行补偿
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        // 否则的话，当前的workerCount < corePoolSize，就需要添加一个worker，
        // 保持线程数在corePoolSize这个数量
        addWorker(null, false);
    }
}

task = getTask()

private Runnable getTask() {
    boolean timedOut = false; // Did the last poll() time out?
    for (;;) {
        // 获取runState
        int c = ctl.get();
        int rs = runStateOf(c);
        // 这个条件的成立为线程池不再是running状态，即 shutdown，stop，TIDYING，TERMINATED
        // 这四种状态，这四种状态，都不允许继续添加新的task，所以直接将workerCount减一
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }
        int wc = workerCountOf(c);
        // allowCoreThreadTimeOut默认为false，
        // 即timed 等于 当前的workerCount是否超过了核心线程数
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        // wc> maximumPoolSize 一般是不会成立的，那么，只有靠超时的的判断
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }
        try {
            // 从 workQueue中取出一个task，根据timed，判断是否需要超时的限制
            // 如果有超时的限制，且未从队列中取出东西来，timeOut就是true
            // 再上述的判断中(timed && timedOut)就会成立,就会返回null，并减少workerCount
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
            workQueue.take();
            // 正常取到任务，返回执行，此时没有减少workerCount的数量，
            // 会在processWorkerExit中减少
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

线程池(Executor)源码浅读

首先，在Java中，创建线程有如下四种方式

线程池

为什么需要线程池

线程池的特点：线程服用，控制最大的并发数，管理线程

何时使用线程池

Java中线程池的实现

Java中线程池的继承图

Java中线程池的分类