首先复习下创建线程的几种方式
1、实现runnable接口
new Thread(() -> log.info("方式一：实现runnable接口")).start();
2、实现callable接口
FutureTask<String> task = new FutureTask<>(() -> "方式二：实现callable接口");
new Thread(task).start();
3、继承Thread类
class MyThread extends Thread{
@Override
public void run() {
log.info("方式三：继承thread类");
}
};
4、使用线程池 这个只是创建了线程池，但是线程具体需要执行的任务还是需要在方法中去指定
ExecutorService executorService = Executors.newCachedThreadPool();
executorService.submit(() -> log.info("方式四：使用线程池"));
ok 现在我们开始来熟悉线程池
1、创建线程池
public ThreadPoolExecutor(int corePoolSize, //核心线程数
int maximumPoolSize, //最大线程数
long keepAliveTime, //存活时间
TimeUnit unit, //时间单位
BlockingQueue<Runnable> workQueue, //阻塞队列，用来储存线程
ThreadFactory threadFactory, //创建线程的工厂
RejectedExecutionHandler handler //拒绝策略 ) {
}
//具体代码引发的问题
public class ThreadPoolListener implements BeanPostProcessor , InitializingBean {
//设置阻塞队列大小为2
BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(2);
//设置核心线程数为4 最大线程数为6
ThreadPoolExecutor pool = new ThreadPoolExecutor(4, 6, 10, TimeUnit.SECONDS, queue, (runnable, executor) -> {
//拒绝策略需要处理的业务逻辑
log.info("do something ~");
log.info(runnable.toString());
log.info(executor.toString());
});
//定时任务处理业务~~~
@Scheduled(cron = "*/5 * * * * *")
@Async
public void logCollection(){
//打印线程池的工作中的线程数以及线程id 队列的大小
pool.execute(() ->{
log.info("测试线程池工作原理 线程数:{},当前线程id:{}",pool.getPoolSize(),Thread.currentThread().getId());
log.info(queue.size()+"----->>队列大小");
if(queue.size()>0){
log.info("队列中的数据:{}",queue.peek());
}
try {
Thread.sleep(100000);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
@Override
public void afterPropertiesSet() throws Exception {
log.info("属性设置完毕 队列容量:{}",queue.size());
}
}
从图中可以看到，在线程数量超过了核心线程之后，队列立马就满了 这是为什么?
从execute入手，看看他都做了什么事情
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
//ctl:原子计数器，来保存当前的正在执行的线程数以及线程状态（高3位存状态，低29位存线程数）
int c = ctl.get();
//执行中的线程数是否小于核心线程数
if (workerCountOf(c) < corePoolSize) {
//添加一个worker来处理任务
if (addWorker(command, true))
return;
//走到这里说明worker添加失败了 再次获取下线程的状态数
c = ctl.get();
}
//如果线程池是运行的状态，将本任务加入到队列中
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
//再判断线程池是不是运行中 不是的话，将本线程移出队列
if (! isRunning(recheck) && remove(command))
// 然后执行拒绝策略
reject(command);
//线程数量为0的话， 开启创建一个新的worker 注意传的参数
else if (workerCountOf(recheck) == 0)
//开启一个新worker，去处理队列中的任务 这里false，在方法里面会作为判断允许线程的数量的条件
//上面一行话看不懂，可以去看看这个方法，就自然懂了
addWorker(null, false);
}
//线程池不是运行中 或者任务加入队列失败 再次尝试添加一个worker去处理任务
else if (!addWorker(command, false))
//添加失败，执行拒绝策略
reject(command);
}
从上面我们可以看到，在判断完线程池的状态以及数量之后，要么执行拒绝策略，要么创建一个worker，要么将任务加入到队列中，所以具体是怎么去处理线程任务？我们来分析addWorker
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
//如果线程已经关闭 并且 任务为null，队列不为空 直接返回
//（因为线程已经关闭，任务队列也为空，当前也没有需要执行的任务。所以直接返回）
//(如果任务队列不为空，或者当前需要执行的任务不为null，并且线程池状态为shutdown，
//线程池会执行完当前的任务，如果状态为stop，则直接返回，所有任务都会被抛弃)
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);
//当前线程数量线程最大允许数量 直接返回
if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//尝试增加线程数量
if (compareAndIncrementWorkerCount(c))
//成功的话，跳出外层循环
break retry;
//走到这里说明由于并发导致线程数量增加失败
c = ctl.get();
//再次判断下线程池状态 如果状态被改变了，那么执行下一次的外层循环 在去获取线程池状态、数量等信息
if (runStateOf(c) != rs)
continue retry;
}
}
//到了这里，说明已经成功增加了线程数量，才会跳出上面的循环
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//增加一个新的worker，将我们传过来的任务传进去，作为线程的第一个任务去执行
//worker中会创建一个线程，一个worker对应一个线程
w = new Worker(firstTask);
//拿到worker中的线程
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
//加独占锁 这一块就与aqs关联起来了 怎么加锁，查看我前面的文章
mainLock.lock();
try {
int rs = runStateOf(ctl.get());
//线程池是运行中的状态 或者是shutdown，并且本任务不为null
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
//本线程已经在运转，直接抛异常
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
//将这个worker加入到工作组中
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
//释放锁 看到这里，我们可以知道，
//上面枷锁，只是为了保证在添加woker到工作组中，线程池的状态不被改变
mainLock.unlock();
}
//worker被成功加入到工作组中了。再来启动这个worker去工作
if (workerAdded) {
//重点！！！
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
//worker没启动，需要做一些清理工作，如前面 workCount 加了 1，将其减掉
addWorkerFailed(w);
}
//返回worker的工作状态
return workerStarted;
}
我们都知道，在调用了thread.start()方法之后，他会启动线程，然后去执行线程的run方法。同样，我们在将worker启动之后，也会调用他的run方法。我们来看run里面都做了什么
这里做一个简单说明：在创建worker的时候，他们将我们需要执行的任务作为必要的构造器参数传进去，在构造方法里面，会对他做一个包装之后，在赋给worker中的thread字段。
public void run() {
runWorker(this);
}
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
//获取第一个任务
Runnable task = w.firstTask;
//将第一个任务置空
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
//第一个任务不为null 或者为null，那么就去取任务
while (task != null || (task = getTask()) != null) {
//加锁
w.lock();
//线程池状态为stop，那么本线程直接中断进行
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 {
//到这里有两种情况
//1 任务为空 那么正常走到这里completedAbruptly 已被改为false
//2 出现了异常 此时completedAbruptly由于异常，所以他的值还是true
processWorkerExit(w, completedAbruptly);
}
}
任务为空 或者出现异常之后，来在这里做处理
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;
//将本worker移除工作组
workers.remove(w);
} finally {
mainLock.unlock();
}
//更改线程池状态 略过吧 不是重点 感兴趣的可以自己看 也很简单 反正我没看
tryTerminate();
int c = ctl.get();
//如果线程池的状态<stop
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);
}
}
再来看看 线程是如何获取任务的
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
//线程池状态>=stop 或者线程池状态=shutdown，队列为空 也是再次保证，线程shutdown之后，不在接收新的创建线程然后去处理任务的请求 但是已经存在的任务还是可以正常去处理
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
//减少线程的数量
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
//正在工作的线程是否超过核心线程数
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
//队列为空 或者线程池状态改变 那么也返回null
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) {
// 如果此 worker 发生了中断，采取的方案是重试
// 解释下为什么会发生中断，这个读者要去看 setMaximumPoolSize 方法。
// 如果开发者将 maximumPoolSize 调小了，导致其小于当前的 workers 数量，
// 那么意味着超出的部分线程要被关闭。重新进入 for 循环，自然会有部分线程会返回 null
timedOut = false;
}
}
}
至此，关于任务是怎么加入到队列，以及又是怎么被取出来然后去执行的已经很清晰了
其他一些细节点
corePoolSize
核心线程数，不要抠字眼，反正先记着有这么个属性就可以了。
maximumPoolSize
最大线程数，线程池允许创建的最大线程数。
workQueue
任务队列，BlockingQueue 接口的某个实现（常使用 ArrayBlockingQueue 和 LinkedBlockingQueue）。
keepAliveTime
空闲线程的保活时间，如果某线程的空闲时间超过这个值都没有任务给它做，那么可以被关闭了。注意这个值并不会对所有线程起作用，如果线程池中的线程数少于等于核心线程数 corePoolSize，那么这些线程不会因为空闲太长时间而被关闭，当然，也可以通过调用 allowCoreThreadTimeOut(true)使核心线程数内的线程也可以被回收。
threadFactory
用于生成线程，一般我们可以用默认的就可以了。通常，我们可以通过它将我们的线程的名字设置得比较可读一些，如 Message-Thread-1， Message-Thread-2 类似这样。
handler：
当线程池已经满了，但是又有新的任务提交的时候，该采取什么策略由这个来指定。有几种方式可供选择，像抛出异常、直接拒绝然后返回等，也可以自己实现相应的接口实现自己的逻辑，这个之后再说。
参考文章：https://javadoop.com/post/java-thread-pool#toc_4