java官方提供了一个Executors来帮助我们创建线程池 但是不推荐 因为这样对线程的控制粒度比较低 我们只能传入创建的线程数量 并不能知道线程的名字和很多东西 推荐手动创建线程池
创建线程池的各种参数：
public ThreadPoolExecutor(int corePoolSize, 核心线程数
int maximumPoolSize, 最大线程数
long keepAliveTime, 最大空闲时间
TimeUnit unit, 时间单位
BlockingQueue<Runnable> workQueue, 阻塞队列
ThreadFactory threadFactory, 线程工厂 主要可以指定创建工厂的名字
RejectedExecutionHandler handler) 拒绝策略
线程池中的核心属性：
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); 这个是一个状态位标志我们的线程池状态
高三位代表状态 低29位代表线程池的数量 这个AtomicInteger就是int类型 但是它对int类型修改时做了CAS操作
private static final int COUNT_BITS = Integer.SIZE - 3; 这个数就是29 因为int类型是32位的
private static final int CAPACITY = (1 << COUNT_BITS) - 1; 其实就是前三位是0 后29位全是1
// 线程池的状态
private static final int RUNNING = -1 << COUNT_BITS; 正在运行状态 前三位 111
private static final int SHUTDOWN = 0 << COUNT_BITS; 表示shutdown状态 前三位000 不接收新线程 但是会处理阻塞队列中的任务 对正在执行的任务也正常处理
private static final int STOP = 1 << COUNT_BITS; 前三位是001 不接收新线程 也不去处理阻塞队列中的任务 同时会中断正在执行的任务
private static final int TIDYING = 2 << COUNT_BITS; 010 表示线程池是一个即将要销毁的中间态
private static final int TERMINATED = 3 << COUNT_BITS; 011 表示线程池已经销毁
// Packing and unpacking ctl
private static int runStateOf(int c) { return c & ~CAPACITY; } 得到线程池当前的状态信息
private static int workerCountOf(int c) { return c & CAPACITY; } 得到当前线程池的线程数量
线程池的执行流程：
线程池的状态变化：
线程池的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;
c = ctl.get();//创建失败证明有并发操作导致创建失败 这里可能是线程池的状态被修改了，或者其他线程抢先创建工作线程 导致我们的线程数量达到了额定的核心线程数 所以重新获取以下标志位
}
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);
}
else if (!addWorker(command, false))//如果是运行状态 但是放入阻塞队列失败 但是核心线程已经满了 试一试创建非核心线程来处理它
reject(command);//如果失败 执行拒绝策略
}
接下来我们来看一看添加工作线程是如何执行的：
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {//首先为外层for循环命名 retry
int c = ctl.get();//获取标志的状态位
int rs = runStateOf(c);//获取线程池当前状态
// 如果线程池不是Running状态 同时 不是！！下述情况： 是SHUTDOWN状态，没有传入任务，阻塞队列不空 就是我们的上面的这个里面的：else if (workerCountOf(recheck) == 0)。 如果同时满足这两种 证明没有处在执行任务的状态 直接返回false 标志工作创建失败
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);//查看当前正在工作的线程数
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))//如果超越了最大线程数 或者 当传入true代表是创建核心线程，大于规定的核心线程数 false创建非核心线程，大于最大线程数 那么创建失败 无法创建任务
return false;
if (compareAndIncrementWorkerCount(c))//CAS操作 将c+1，目的是将线程数+1
break retry;//如果成功就跳出外层循环
c = ctl.get(); //每成功证明有并发操作 修改了状态位 或者创建了新线程 让数量达到最大 重新获取一些 标志位
if (runStateOf(c) != rs)//如果和上次的运行状态不同 那么证明修改了运行状态 重新进入下次外层循环去检查运行状态
continue retry;
//如果相同则证明应该是线程数量发生改变 不用从外层循环开始继续 直接继续循环添加线程
}
}
boolean workerStarted = false;//工作线程是开始工作
boolean workerAdded = false;//工作是否已经添加 但是未启动
Worker w = null;
try {
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());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {//如果是运行状态或者 是SHUTDOWN状态 而且 任务为空 又是这个语句里面的可能： else if (workerCountOf(recheck) == 0)
if (t.isAlive()) //检查线程是否是存活的 如果没有抛异常
throw new IllegalThreadStateException();
workers.add(w);//将我们的work加入线程池的工作队列
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);//添加到失败队列中
}
return workerStarted;//返回工作是否开始执行的标志位
}
我们再看一看Worker对象是如何封装的：
Worker(Runnable firstTask) {//构造方法 将任务赋值
setState(-1); //这个好像是AbstractQueuedSynchronizer里面的内容 之后去了解
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);//从工厂中新建一个线程同时将自己作为参数传入，那么调用 thread的start方法 就会执行我们的Worker的run方法
}
我们看一看run方法，观察一下线程调用了start方法过后的流程：
调用runWorker方法 将自己作为参数传入：
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();//获取当前的线程 就是我们调用start来执行这个任务的线程 不是主线程
Runnable task = w.firstTask;//将任务获取到
w.firstTask = null;//这两个好像是AQS的工作 之后去看
w.unlock();
boolean completedAbruptly = true;//标志位
try {
//当任务不为空 或者任务为空的话 就从阻塞队列中去获取 这个getTask是一个阻塞操作 没有获取到之前都会阻塞在这里 之后需要去看的方法getTask()
while (task != null || (task = getTask()) != null) {
w.lock();//上锁 防止并发
//如果当前状态的标志位大于等于STOP 意思是如果当前线程池的状态为 STOP TERMINATED TYDING状态下 而且线程还不是中断状态 那么就对这个执行任务的线程进行中断操作 不要让他再继续执行任务了 因为这几个标志位就是不要执行任务的标志位
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
beforeExecute(wt, task);//类似AOP操作的执行前的操作 默认是空方法 需要我们自定义实现
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);//类似AOP操作的执行前的操作 默认是空方法 需要我们自定义实现
}
} finally {
task = null;
w.completedTasks++;
w.unlock();//解锁
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);//这个方法需要查看
}
}
下面再展示一下如何实现beforeExecute和afterExecute其实很简单：
private static class MyThreadPool extends ThreadPoolExecutor{
public MyThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
}
public MyThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
}
public MyThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
}
public MyThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
}
@Override
protected void beforeExecute(Thread t, Runnable r) {
System.out.println("执行任务前的操作");
}
@Override
protected void afterExecute(Runnable r, Throwable t) {
System.out.println("执行任务后的操作");
}
}
public static void main(String[] args) {
ThreadPoolExecutor threadPoolExecutor = new MyThreadPool(2,
4,
10,
TimeUnit.MILLISECONDS,
new PriorityBlockingQueue<>(2),
new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r);
return thread;
}
}, new ThreadPoolExecutor.AbortPolicy());
threadPoolExecutor.execute(()->{
for (int i=0;i<2;i++){
System.out.println(-1 << 29);
}
});
}
执行结果：
