多线程五 线程池 ThreadPoolExecutor
目录什么是线程池线程池的优势Java 中提供的线程池 APIExcutorsThreadpoolExecutor线程池初始化以后做了什么事情newFixedThreadPoolnewCachedThreadPoolnewSingleThreadExecutor线程池原理分析(FixedThreadPool)源码分析execute状...
目录
什么是线程池
线程池的核心逻辑是提前创建好若干个线程放在一个容器中。如果有任务需要处理，则将任务直接分配给线程池中的线程来执行就行，任务处理完以后这个线程不会立刻被销毁，而是等待后续分配任务。同时通过线程池来重复管理线程还 可以避免创建大量线程增加开销。
线程池的优势
1 复用已有的资源
---》降低创建线程和销毁线程的性能开销
---》提高响应速度，当有新任务需要执行是不需要等待线程创建就可以立马执行
2 控制资源总数
--》合理的设置线程池的大小可以避免因为线程数超过硬件资源瓶颈带来的问题
Java 中提供的线程池 API
demo
public class Test implements Runnable{
@Override
public void run() {
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName());
}
static ExecutorService service=Executors.newFixedThreadPool(3);
public static void main(String[] args) {
for(int i=0;i<100;i++) {
service.execute(new Test());
}
service.shutdown();
}
} Excutors
提供了几个线程池的工厂方法
newFixedThreadPool：该方法返回一个固定数量的线程池，线程数不变，当有一个任务提交 时，若线程池中空闲，则立即执行，若没有，则会被暂缓在一个任务队列中，等待有空闲的 线程去执行。
newSingleThreadExecutor: 创建一个线程的线程池，若空闲则执行，若没有空闲线程则暂缓 在任务队列中。
newCachedThreadPool：返回一个可根据实际情况调整线程个数的线程池，不限制最大线程数量，若用空闲的线程则执行任务，若无任务则不创建线程。并且每一个空闲线程会在60秒 后自动回收
newScheduledThreadPool: 创建一个可以指定线程的数量的线程池，但是这个线程池还带有延迟和周期性执行任务的功能，类似定时器。
ThreadpoolExecutor
上面提到的四种线程池的构建，都是基于ThreadpoolExecutor来构建的
public ThreadPoolExecutor(int corePoolSize, //核心线程数量
int maximumPoolSize, //最大线程数
long keepAliveTime, //超时时间,超出核心线程数量以外的线程空余存活时间
TimeUnit unit, //存活时间单位
BlockingQueue<Runnable> workQueue, //保存执行任务的队列
ThreadFactory threadFactory,//创建新线程使用的工厂
RejectedExecutionHandler handler //当任务无法执行的时候的处理方式) 线程池初始化以后做了什么事情
线程池初始化时是没有创建线程的，线程池里的线程的初始化与其他线程一样，但是在完成任务以后，该线程不会自行销毁，而是以挂起的状态返回到线程池。直到应用程序再次向线 程池发出请求时，线程池里挂起的线程就会再度激活执行任务。这样既节省了建立线程所造 成的性能损耗，也可以让多个任务反复重用同一线程，从而在应用程序生存期内节约大量开销
newFixedThreadPool
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}FixedThreadPool 的核心线程数和最大线程数都是指定值，也就是说当线程池中的线程数超 过核心线程数后，任务都会被放到阻塞队列中。另外 keepAliveTime 为 0，也就是超出核心线程数量以外的线程空余存活时间为0，而这里选用的阻塞队列LinkedBlockingQueue，使用的是默认容量 Integer.MAX_VALUE， 相当于没有上限 。
这个线程池执行任务的流程如下：
1. 线程数少于核心线程数，也就是设置的线程数时，新建线程执行任务
2. 线程数等于核心线程数后，将任务加入阻塞队列
3. 由于队列容量非常大，可以一直添加
4. 执行完任务的线程反复去队列中取任务执行
用途：FixedThreadPool 用于负载比较大的服务器，为了资源的合理利用，需要限制当前线程数量
newCachedThreadPool
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}CachedThreadPool 创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空 闲线程，若无可回收，则新建线程; 并且没有核心线程，非核心线程数无上限，但是每个空闲的时间只有60秒，超过后就会被回收。
它的执行流程如下：
1. 没有核心线程，直接向 SynchronousQueue 中提交任务
2. 如果有空闲线程，就去取出任务执行；如果没有空闲线程，就新建一个
3. 执行完任务的线程有 60 秒生存时间，如果在这个时间内可以接到新任务，就可以继续活 下去，否则就被回收
缺点：因为它的 阻塞队列是SynchronousQueue 是不存储元素的阻塞队列，最大值是在初始化的时候设置为 Integer.MAX_VALUE，可能短时间内创建出大量的非核心线程容易造成堆外内存溢出。
newSingleThreadExecutor
创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定 顺序(FIFO, LIFO, 优先级)执行
线程池原理分析(FixedThreadPool)
线程池创建的线程存储在了HashSet<Worker> 中
源码分析
execute
基于源码入口进行分析，先看execute方法
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
//1.当前池中线程比核心数少，新建一个线程执行任务
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
//2.核心池已满，但任务队列未满，添加到队列中
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);
}
//3.核心池已满，队列已满，试着创建一个新线程
else if (!addWorker(command, false))
//如果创建新线程失败了，说明线程池被关闭或者线程池完全满了，拒绝任务
reject(command);
} 状态转化 ctl 的作用
在线程池中，ctl贯穿在线程池的整个生命周期中
ctl：private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
它是一个原子类，主要作用是用来保存线程数量和线程池的状态。一个int数值是4个字节32个bit位，这里采用高3位来保存运行状态，低29位来保存线程数量。
原理是如何实现的呢：
第一个问题是一个变量如何存储两个数据：
首先，int类型是4个字节，也就是32位，例如一个int值在计算机中的表示：00000000 01000100 11111111 00000000
因为ThreadPoolExecutor中定义的状态有5种（RUNNING、SHUTDOWN、STOP、TIDYING、TERMINATED）
用1位表示，则只有0、1两种情况，只能表示两种状态；
用2位表示，有00、01、10、11四种情况，只能表示四种状态；
用3位表示，有222=8种情况；
所以状态值至少要用3位，那么就可以用int的高3位来表示（最左边3个），剩下29个就可以表示线程数量（所以线程数量最大值就是29位上全是1）。
每当线程池中的线程数量或状态发生变化时，具体操作的便是ctl变量，如以下方法：
private boolean compareAndIncrementWorkerCount(int expect) {
return ctl.compareAndSet(expect, expect + 1);
}
那么又是读取线程状态和数量的值呢：
读取状态利用以下方法：
private static int runStateOf(int c) { return c & ~CAPACITY; }
CAPACITY是个常量00011111 11111111 11111111 11111111，通过 &（按位与）运算，可以保留高3位，把低29位全部变为0；
读取数量利用以下方法：
private static int workerCountOf(int c) { return c & CAPACITY; }
可以把高3位变为0，低29位保留。
其中传入的参数c,就是ctl。
private static final int COUNT_BITS = Integer.SIZE - 3; //32-3 =29
private static final int CAPACITY = (1 << COUNT_BITS) - 1; //将 1 的二进制向右位移 29 位,再减 1 表示最大线程容量
// 运行状态保存在 int值的高 3位 ( 所有数值左移 29位 )
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; //所有的任务都已结束,线程数量为 0,处于该状态的线程池即将调用 terminated()方法
private static final int TERMINATED = 3 << COUNT_BITS;// terminated()方法执行完成
addWorker
如果工作线程数小于核心线程数的话，会调用addWorker，顾名思义，其实就是要创建一个 工作线程。
源码比较长，看起来比较唬人,其实就做了两件事。
1）才用循环 CAS 操作来将线程数加 1；
2）新建一个线程并启用。
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.
如果线程处于非运行状态，并且 rs不等于 SHUTDOWN且 firstTask不等于空且且workQueue为空，直接返回 false （表示不可添加 work状态）
1. 线程池已经 shutdown后，还要添加新的任务，拒绝
2. （第二个判断） SHUTDOWN状态不接受新任务，但仍然会执行已经加入任务队列的任务,
所以当进入 SHUTDOWN状态，而传进来的任务为空，并且任务队列不为空的时候，是允许添加新线程的 , 如果把这个条件取反，就表示不允许添加 worker
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) { //自旋
int wc = workerCountOf(c);//获得 Worker 工作线程数
//如果工作线程数大于默认容量大小或者大于核心线程数大小，则直接返回 false 表示不能再添加 worker。
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//通过 cas 来增加工作线程数，如果 cas 失败，则直接重试
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl // 再次获取 ctl
的值
if (runStateOf(c) != rs) //这里如果不想等，说明线程的状态发生了变化，
继续重试
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
//上面这段代码主要是对 worker 数量做原子+1 操作,下面的逻辑才是正式构建一个 worker
boolean workerStarted = false; //工作线程是否启动的标识
boolean workerAdded = false; //工作线程是否已经添加成功的标识
Worker w = null;
try {
w = new Worker(firstTask); //构建一个 Worker，这个 worker 是什么呢？我们
可以看到构造方法里面传入了一个 Runnable 对象
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 且 firstTask 为空]，才
能添加到 workers 集合中
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
//任务刚封装到 work 里面，还没 start,你封装的线程就是 alive，几
个意思？肯定是要抛异常出去的
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
workers.add(w); //将新创建的 Worker 添加到 workers 集合中
int s = workers.size();
//如果集合中的工作线程数大于最大线程数，这个最大线程数表示线程池曾经出现过的最大线程数
if (s > largestPoolSize)
largestPoolSize = s; //更新线程池出现过的最大线程数
workerAdded = true;//表示工作线程创建成功了
}
} finally {
mainLock.unlock(); //释放锁
}
if (workerAdded) {//如果 worker 添加成功
t.start();//启动线程
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w); //如果添加失败，就需要做一件事，就是递减实际工作线
程数(还记得我们最开始的时候增加了工作线程数吗)
}
return workerStarted;//返回结果
}
Worker 类说明
我们发现addWorker方法只是构造了一个Worker，并且把firstTask封装到worker中，它是 做什么的呢？我们来看看
1. 每个worker,都是一条线程,同时里面包含了一个firstTask,即初始化时要被首先执行的任务.
2. 最终执行任务的,是runWorker()方法
Worker类继承了AQS，并实现了Runnable接口，注意其中的firstTask和thread属性： firstTask用它来保存传入的任务；thread是在调用构造方法时通过ThreadFactory来创建的 线程，是用来处理任务的线程。
在调用构造方法时，需要传入任务，这里通过getThreadFactory().newThread(this);来新建 一个线程，newThread方法传入的参数是this，因为Worker本身继承了Runnable接口， 也就是一个线程，所以一个Worker对象在启动的时候会调用Worker类中的run方法。
Worker继承了AQS，使用AQS来实现独占锁的功能。为什么不使用ReentrantLock来实 现呢？可以看到tryAcquire方法，它是不允许重入的，而ReentrantLock是允许重入的；
- lock方法一旦获取了独占锁，表示当前线程正在执行任务中；那么它会有以下几个作用
- 如果正在执行任务，则不应该中断线程；
- 如果该线程现在不是独占锁的状态，也就是空闲的状态，说明它没有在处理任务，这时可 以对该线程进行中断；
- 线程池在执行shutdown方法或tryTerminate方法时会调用interruptIdleWorkers方法来 中断空闲的线程，interruptIdleWorkers方法会使用tryLock方法来判断线程池中的线程 是否是空闲状态
- 之所以设置为不可重入，是因为我们不希望任务在调用像setCorePoolSize这样的线程池 控制方法时重新获取锁，这样会中断正在运行的线程
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
private static final long serialVersionUID = 6138294804551838833L;
/** Thread this worker is running in. Null if factory fails. */
final Thread thread; //注意了，这才是真正执行 task 的线程，从构造函数可知是由
ThreadFactury 创建的
/** Initial task to run. Possibly null. */
Runnable firstTask; //这就是需要执行的 task
/** Per-thread task counter */
volatile long completedTasks; //完成的任务数，用于线程池统计
Worker(Runnable firstTask) {
// 初始状态 -1, 防止在调用 runWorker() ，也就是真正执行task前中断 thread 。
setState(-1);
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
public void run() {
runWorker(this);
}
protected boolean isHeldExclusively() {
return getState() != 0;
}
protected boolean tryAcquire(int unused) {
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
protected boolean tryRelease(int unused) {
setExclusiveOwnerThread(null);
setState(0);
return true;
}
public void lock() { acquire(1); }
public boolean tryLock() { return tryAcquire(1); }
public void unlock() { release(1); }
public boolean isLocked() { return isHeldExclusively(); }
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null
&& !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
} runWorker 方法
前面已经了解了 ThreadPoolExecutor 的核心方法 addWorker，主要作用是增加工作线程， 而 Worker 简单理解其实就是一个线程，里面重新了 run 方法，这块是线程池中执行任务的 真正处理逻辑，也就是runWorker方法，这个方法主要做几件事
- 1. 如果task不为空,则开始执行task
- 2. 如果task为空,则通过getTask()再去取任务,并赋值给task,如果取到的Runnable不为空,则 执行该任务
- 3. 执行完毕后,通过while循环继续getTask()取任务
- 4. 如果getTask()取到的任务依然是空,那么整个runWorker()方法执行完毕
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
unlock，表示当前 worker 线程允许中断，因为 new Worker 默认的 state=-1,此处是调用
Worker 类的 tryRelease()方法，将 state 置为 0，而 interruptIfStarted()中只有 state>=0 才允许调用中断
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
//注意这个 while 循环,在这里实现了 [线程复用] // 如果 task 为空，则通过
getTask 来获取任务
while (task != null || (task = getTask()) != null) {
w.lock(); //上锁，不是为了防止并发执行任务，为了在 shutdown()时不终止正
在运行的 worker
线程池为 stop
状态时不接受新任务，不执行已经加入任务队列的任务，还中断正在执
行的任务
// 所以对于 stop
状态以上是要中断线程的
//(Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP) 确保线
程中断标志位为 true
且是 stop
状态以上，接着清除了中断标志
//!wt.isInterrupted() 则再一次检查保证线程需要设置中断标志位
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
beforeExecute(wt, task);//这里默认是没有实现的，在一些特定的场景中
我们可以自己继承 ThreadpoolExecutor 自己重写
Throwable thrown = null;
try {
task.run(); //执行任务中的 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,需要再通过 getTask()
取) + 记录该 Worker 完成任务数量 + 解锁
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
//1.将入参 worker 从数组 workers 里删除掉；
//2.根据布尔值 allowCoreThreadTimeOut 来决定是否补充新的 Worker 进数组
workers
}
}getTask 如何判断出空闲线程超时
1 worker线程会从阻塞队列中获取需要执行的任务
2 怎样判断线程有多久没有活动了
怎样判断线程有多久没有活动了，是不是以为 线程池 会启动一个监控线程，专 门监控哪个线程正在偷懒？想太多，其实只是在线程从工作队列 poll 任务时，加上了超时 限制，如果线程在 keepAliveTime 的时间内 poll 不到任务，那我就认为这条线程没事做， 可以干掉了，看看这个代码片段你就清楚了
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {//自旋
int c = ctl.get();
int rs = runStateOf(c);
* 对线程池状态的判断，两种情况会 workerCount-1，并且返回 null
1. 线程池状态为 shutdown，且 workQueue 为空（反映了 shutdown 状态的线程池还是
要执行 workQueue 中剩余的任务的）
2. 线程池状态为 stop（shutdownNow()会导致变成 STOP）（此时不用考虑 workQueue
的情况）
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;//返回 null，则当前 worker 线程会退出
}
int wc = workerCountOf(c);
// timed 变量用于判断是否需要进行超时控制。
// allowCoreThreadTimeOut 默认是 false，也就是核心线程不允许进行超时；
// wc > corePoolSize，表示当前线程池中的线程数量大于核心线程数量；
// 对于超过核心线程数量的这些线程，需要进行超时控制
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
1. 线程数量超过 maximumPoolSize 可能是线程池在运行时被调用了 setMaximumPoolSize()
被改变了大小，否则已经 addWorker()成功不会超过 maximumPoolSize
2. timed && timedOut 如果为 true，表示当前操作需要进行超时控制，并且上次从阻塞队列中
获取任务发生了超时.其实就是体现了空闲线程的存活时间
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
根据 timed 来判断，如果为 true，则通过阻塞队列 poll 方法进行超时控制，如果在
keepaliveTime 时间内没有获取到任务，则返回 null.
否则通过 take 方法阻塞式获取队列中的任务
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)//如果拿到的任务不为空，则直接返回给 worker 进行处理
return r;
timedOut = true;//如果 r==null，说明已经超时了，设置 timedOut=true，在
下次自旋的时候进行回收
} catch (InterruptedException retry) {
timedOut = false; // 如果获取任务时当前线程发生了中断，则设置 timedOut 为
false 并返回循环重试
}
} } execute 后续逻辑分析
如果核心线程数已满，说明这个时候不能再创建核心线程了，于是走第二个判断
第二个判断逻辑比较简单，如果线程池处于运行状态并且任务队列没有满，则将任务添加到队列中
第三个判断，核心线程数满了，队列也满了，那么这个时候创建新的线程也就是（非核心线 程） 如果非核心线程数也达到了最大线程数大小，则直接拒绝任务
//2.核心池已满，但任务队列未满，添加到队列中
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);
}
//3.核心池已满，队列已满，试着创建一个新 线程
else if (!addWorker(command, false))
//如果创建新线程失败了，说明线程池被关闭或者线程池完全满 了，拒绝任务
reject(command); 拒绝策略
线程池的拒绝策略，是指当任务添加到线程池中被拒绝，而采取的处理措施。
当任务添加到线程池中之所以被拒绝，可能是由于：第一，线程池异常关闭。第二，任务数量超过线程池的最大限制。
线程池共包括4种拒绝策略，它们分别是：AbortPolicy, CallerRunsPolicy, DiscardOldestPolicy和DiscardPolicy。
AbortPolicy -- 当任务添加到线程池中被拒绝时，它将抛出 RejectedExecutionException 异常。 CallerRunsPolicy -- 当任务添加到线程池中被拒绝时，会在线程池当前正在运行的Thread线程池中处理被拒绝的任务。 DiscardOldestPolicy -- 当任务添加到线程池中被拒绝时，线程池会放弃等待队列中最旧的未处理任务，然后将被拒绝的任务添加到等待队列中。 DiscardPolicy -- 当任务添加到线程池中被拒绝时，线程池将丢弃被拒绝的任务。
当然也可以根据应用场景实现RejectedExecutionHandler接口，自定义饱和策略，如记录 日志或持久化存储不能处理的任务
线程池默认的处理策略是AbortPolicy！
DiscardPolicy 示例
import java.lang.reflect.Field;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ThreadPoolExecutor.DiscardPolicy;
public class DiscardPolicyDemo {
private static final int THREADS_SIZE = 1;
private static final int CAPACITY = 1;
public static void main(String[] args) throws Exception {
// 创建线程池。线程池的"最大池大小"和"核心池大小"都为1(THREADS_SIZE)，"线程池"的阻塞队列容量为1(CAPACITY)。
ThreadPoolExecutor pool = new ThreadPoolExecutor(THREADS_SIZE, THREADS_SIZE, 0, TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(CAPACITY));
// 设置线程池的拒绝策略为"丢弃"
pool.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());
// 新建10个任务，并将它们添加到线程池中。
for (int i = 0; i < 10; i++) {
Runnable myrun = new MyRunnable("task-"+i);
pool.execute(myrun);
}
// 关闭线程池
pool.shutdown();
}
}
class MyRunnable implements Runnable {
private String name;
public MyRunnable(String name) {
this.name = name;
}
@Override
public void run() {
try {
System.out.println(this.name + " is running.");
Thread.sleep(100);
} catch (Exception e) {
e.printStackTrace();
}
}
}运行结果：
task-0 is running.
task-1 is running.结果说明：线程池pool的"最大池大小"和"核心池大小"都为1(THREADS_SIZE)，这意味着"线程池能同时运行的任务数量最大只能是1"。
线程池pool的阻塞队列是ArrayBlockingQueue，ArrayBlockingQueue是一个有界的阻塞队列，ArrayBlockingQueue的容量为1。这也意味着线程池的阻塞队列只能有一个线程池阻塞等待。
根据""中分析的execute()代码可知：线程池中共运行了2个任务。第1个任务直接放到Worker中，通过线程去执行；第2个任务放到阻塞队列中等待。其他的任务都被丢弃了！
DiscardOldestPolicy 示例
public class DiscardOldestPolicyDemo {
private static final int THREADS_SIZE = 1;
private static final int CAPACITY = 1;
public static void main(String[] args) throws Exception {
// 创建线程池。线程池的"最大池大小"和"核心池大小"都为1(THREADS_SIZE)，"线程池"的阻塞队列容量为1(CAPACITY)。
ThreadPoolExecutor pool = new ThreadPoolExecutor(THREADS_SIZE, THREADS_SIZE, 0, TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(CAPACITY));
// 设置线程池的拒绝策略为"DiscardOldestPolicy"
pool.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());
// 新建10个任务，并将它们添加到线程池中。
for (int i = 0; i < 10; i++) {
Runnable myrun = new MyRunnable("task-"+i);
pool.execute(myrun);
}
// 关闭线程池
pool.shutdown();
}
}
class MyRunnable implements Runnable {
private String name;
public MyRunnable(String name) {
this.name = name;
}
@Override
public void run() {
try {
System.out.println(this.name + " is running.");
Thread.sleep(200);
} catch (Exception e) {
e.printStackTrace();
}
}
}
运行结果：
task-0 is running.
task-9 is running.
结果说明：将"线程池的拒绝策略"由DiscardPolicy修改为DiscardOldestPolicy之后，当有任务添加到线程池被拒绝时，线程池会丢弃阻塞队列中末尾的任务，然后将被拒绝的任务添加到末尾。
线程池的注意事项
阿里开发手册不建议使用Executors去创建
手册上是说 线程池的构建不允许使用Executors去创建，而是通过ThreadPoolExecutor的方式。 分析完原理以后，大家自己一定要有一个答案。我来简单分析下，
用 Executors 使得用户不 需要关心线程池的参数配置，意味着大家对于线程池的运行规则也会慢慢的忽略。这会导致 一个问题，比如我们用 newFixdThreadPool 或者 singleThreadPool.允许的队列长度为 Integer.MAX_VALUE，如果使用不当会导致大量请求堆积到队列中导致OOM的风险 而 newCachedThreadPool，允许创建线程数量为 Integer.MAX_VALUE，也可能会导致大量 线程的创建出现CPU使用过高或者OOM的问题
如何合理配置线程池的大小
在遇到这类问题时，先冷静下来分析
1. 需要分析线程池执行的任务的特性： CPU密集型还是IO密集型
2. 每个任务执行的平均时长大概是多少，这个任务的执行时长可能还跟任务处理逻辑是否涉 及到网络传输以及底层系统资源依赖有关系
如果是CPU密集型，主要是执行计算任务，响应时间很快，cpu一直在运行，这种任务cpu 的利用率很高，那么线程数的配置应该根据CPU核心数来决定，CPU核心数=最大同时执行 线程数，加入CPU核心数为4，那么服务器最多能同时执行4个线程。过多的线程会导致上 下文切换反而使得效率降低。那线程池的最大线程数可以配置为cpu核心数+1
如果是IO密集型，主要是进行IO操作，执行IO操作的时间较长，这是cpu出于空闲状态， 导致cpu的利用率不高，这种情况下可以增加线程池的大小。这种情况下可以结合线程的等 待时长来做判断，等待时间越高，那么线程数也相对越多。一般可以配置cpu核心数的2倍。 一个公式：线程池设定最佳线程数目 = （（线程池设定的线程等待时间+线程 CPU 时间）/ 线程CPU时间 ）* CPU数目 这个公式的线程cpu时间是预估的程序单个线程在cpu上运行的时间（通常使用loadrunner 测试大量运行次数求出平均值）
线程池中的线程初始化
默认情况下，创建线程池之后，线程池中是没有线程的，需要提交任务之后才会创建线程。
在实际中如果需要线程池创建之后立即创建线程，可以通过以下两个方法办到：
prestartCoreThread()：初始化一个核心线程；
prestartAllCoreThreads()：初始化所有核心线程
ThreadPoolExecutor tpe=(ThreadPoolExecutor)service;
tpe.prestartAllCoreThreads();
线程池的关闭
ThreadPoolExecutor 提供了两个方法，用于线程池的关闭，分别是 shutdown()和 shutdownNow()，其中：
shutdown()：不会立即终止线程池，而是要等所有任务缓存队列中 的任务都执行完后才终止，但再也不会接受新的任务
shutdownNow()：立即终止线程池，并 尝试打断正在执行的任务，并且清空任务缓存队列，返回尚未执行的任务
线程池容量的动态调整
ThreadPoolExecutor 提供了动态调整线程池容量大小的方法：setCorePoolSize()和 setMaximumPoolSize()，
setCorePoolSize：设置核心池大小
setMaximumPoolSize：设置线 程池最大能创建的线程数目大小
任务缓存队列及排队策略 在前面我们多次提到了任务缓存队列，即 workQueue，它用来存放等待执行的任务。
workQueue 的类型为 BlockingQueue，通常可以取下面三种类型：
- 1. ArrayBlockingQueue：基于数组的先进先出队列，此队列创建时必须指定大小；
- 2. LinkedBlockingQueue：基于链表的先进先出队列，如果创建时没有指定此队列大小，则默 认为 Integer.MAX_VALUE；
- 3. SynchronousQueue：这个队列比较特殊，它不会保存提交的任务，而是将直接新建一个 线程来执行新来的任务。
线程池的监控
如果在项目中大规模的使用了线程池，那么必须要有一套监控体系，来指导当前线程池的状 态，当出现问题的时候可以快速定位到问题。而线程池提供了相应的扩展方法，我们通过重 写线程池的beforeExecute、afterExecute和shutdown等方式就可以实现对线程的监控，简 单给大家演示一个案例
public class VincentThreadPool extends ThreadPoolExecutor {
// 保存任务开始执行的时间 , 当任务结束时 , 用任务结束时间减去开始时间计算任务执行时间
private ConcurrentHashMap<String, Date> startTimes;
public VincentThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
this.startTimes=new ConcurrentHashMap<>();
}
@Override
public void shutdown() {
System.out.println("已经执行的任务数：" + this.getCompletedTaskCount() + "," +
"当前活动线程数:" + this.getActiveCount() +
",当前排队线程数:"+this.getQueue().size());
System.out.println();
super.shutdown();
}
// 任务开始之前记录任务开始时间
@Override
protected void beforeExecute(Thread t, Runnable r) {
startTimes.put(String.valueOf(r.hashCode()),new Date());
super.beforeExecute(t, r);
}
@Override
protected void afterExecute(Runnable r, Throwable t) {
Date startDate = startTimes.remove(String.valueOf(r.hashCode()));
Date finishDate = new Date();
long diff = finishDate.getTime() - startDate.getTime();
// 统计任务耗时、初始线程数、核心线程数、正在执行的任务数量、
// 已完成任务数量、任务总数、队列里缓存的任务数量、
// 池中存在的最大线程数、最大允许的线程数、线程空闲时间、线程池是否关闭、线程池是否终止
System.out.print("任务耗时:"+diff+"\n");
System.out.print("初始线程数:"+this.getPoolSize()+"\n");
System.out.print("核心线程数:"+this.getCorePoolSize()+"\n");
System.out.print("正在执行的任务数量:"+this.getActiveCount()+"\n");
System.out.print("已经执行的任务数:"+this.getCompletedTaskCount()+"\n");
System.out.print("任务总数:"+this.getTaskCount()+"\n");
System.out.print("最大允许的线程数:"+this.getMaximumPoolSize()+"\n");
System.out.print("线程空闲时间:"+this.getKeepAliveTime(TimeUnit.MILLISECONDS)+"\n");
System.out.println();
super.afterExecute(r, t);
}
}public class VincentThreadPoolTest implements Runnable{
private static ExecutorService es =new VincentThreadPool(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
@Override
public void run() {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public static void main(String[] args) throws Exception {
for (int i = 0; i < 100; i++) {
es.execute(new VincentThreadPoolTest());
}
es.shutdown();
}
}
CSDN联合极客时间，共同打造面向开发者的精品内容学习社区，助力成长！
更多推荐
2
0- 0
已为社区贡献1条内容
所有评论(0)