学习文章
https://mp.weixin.qq.com/s/Qb3pwewrV0cJqXcYvP00kA
Demo
代码
自定义线程池
线程池中如何做到线程的重复利用？
答：
- 线程池不存在取出线程使用完再归还的操作，因为线程调用start方法后，生命周期就不由父线程控制，线程run方法执行完成后就销毁了。
- 线程池中的线程在run方法中开启循环基于生产者消费者模式获取任务。即若消息队列存在任务，则获取执行，若任务为空，则阻塞。其中核心线程在没任务的时候会阻塞不被销毁，等待新任务到来。工作线程(非核心线程)会在run方法执行后自动销毁。
备注：
- 生产者：
ThreadPool.execute:若 corePoolSize 已满且queueSize未满，则入队任务 - 消费者：
Worker 本质为线程，内部存在无限循环不断检查getTask()是否有任务，有任务就进行消费，getTask()主要内容为若消息队列存在任务，则获取执行；若任务为空且为核心线程，则阻塞；若任务为空且为非核心线程，则结束循环让run方法可以执行完成走自动销毁的道路。 - 线程池的execute功能：开启线程和入队
1). 若 corePoolSize 未满，则开启核心线程
2). 若 corePoolSize 已满且queueSize未满，则入队
3). 若 queueSize 已满且maxNumPoolSize未满，则开启非核心线程
4). 若 maxNumberPoolSize 已满，则拒绝策略生效
package com.sufadi.study.thread;
import java.time.LocalTime;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class MyThreadPool implements Executor {
// 线程池最大核心线程数
private int corePoolSize = 1;
// 线程池最大线程数
private int maxNumPoolSize = 3;
// 获取消息队列的超时时长(ms)
private long keepAliveTime = 100;
// 因天生带阻塞机制，故为生产者与消费者模型中的消息队列：入队(queue.offer)、出队(queue.take)、超时出队(queue.poll)
private BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(3);
// 当前线程数量
private AtomicInteger count = new AtomicInteger(0);
public MyThreadPool() {
System.out.println(LocalTime.now() + ": " + "MyThreadPool corePoolSize:" + corePoolSize + ", maxNumPoolSize:" + maxNumPoolSize
+ ", queue.size:" + queue.size() + ", keepAliveTime(ms):" + keepAliveTime);
}
/**
* 线程池的execute功能：开启线程和入队
* 1. 若 corePoolSize 未满，则开启核心线程
* 2. 若 corePoolSize 已满且queueSize未满，则入队
* 3. 若 queueSize 已满且maxNumPoolSize未满，则开启非核心线程
* 4. 若 maxNumberPoolSize 已满，则拒绝策略生效
*
* 线程池的核心：生产者与消费者模型
*/
@Override
public void execute(Runnable task) {
int curThreadSize = count.get();
// 若 corePoolSize 未满，则开启核心线程
if (curThreadSize < corePoolSize && count.compareAndSet(curThreadSize, curThreadSize + 1)) {
new Worker(task).start();
System.out.println(LocalTime.now() + ": " + "1. 若 corePoolSize 未满，则开启核心线程, 当前线程数量:" + count.get());
return;
}
// 若 corePoolSize 已满且queueSize未满，则入队
// Ps: queue.offer 返回值为true则表示未满
if (queue.offer(task)) {
System.out.println(LocalTime.now() + ": " + "2. 若 corePoolSize 已满且queueSize未满，则[入队] 当前线程数量:" + count.get());
return;
}
// 3. 若 queueSize 已满且maxNumPoolSize未满，则开启非核心线程
if (curThreadSize < maxNumPoolSize && count.compareAndSet(curThreadSize, curThreadSize + 1)) {
new Worker(task).start();
System.out.println(LocalTime.now() + ": " + "3. 若 queueSize 已满且maxNumPoolSize未满，则开启非核心线程, 当前线程数量:" + count.get());
return;
}
System.out.println(LocalTime.now() + ": " + "4. 若 maxNumberPoolSize 已满，则拒绝策略生效, 当前线程数量:" + count.get());
}
/**
* Worker 为消费者, 本质为线程,工作流程：
* 先消费Worker的fistTask后，再循环执行getTask获取任务并消费，获取不到task时，就退出循环，线程销毁。
*/
class Worker extends Thread {
Runnable firstTask;
public Worker(Runnable fistTask) {
this.firstTask = fistTask;
}
@Override
public void run() {
Runnable task = firstTask;
firstTask = null;
while (true) {
try {
// 先消费Worker的fistTask后，再循环执行getTask获取任务并消费，获取不到task时，就退出循环，线程销毁。
if (task != null || (task = getTask()) != null) {
System.out.println(LocalTime.now() + ": " + Thread.currentThread().getName() + ",开始消费线程");
task.run();
} else {
break;
}
} catch (Exception e) {
e.printStackTrace();
} finally {
task = null;
}
}
}
}
/**
* 线程池的核心：生产者与消费者模型 getTask()
* 这里的出队分别使用poll和take的原因是区分核心线程和非核心线程的出队获取方式，目的如下：
* 即非核心线程使用poll得到队列为空，超时后，会返回null，不阻塞，这样线程才有机会run执行完毕被销毁；
* 即核心线程使用take在队列为空的的情况下，会一直阻塞，直到新任务到来，即核心线程不死，除非线程池关闭；
*/
private Runnable getTask() throws InterruptedException {
String threadName = Thread.currentThread().getName();
boolean timeOut = false;
while (true) {
int curThreadSize = count.get();
// 超出核心线程才进入超时逻辑
if (curThreadSize > corePoolSize) {
if (timeOut) {
if (count.compareAndSet(curThreadSize, curThreadSize -1)) {
System.out.println(LocalTime.now() + ": " + threadName + " 已超时, 可以销毁了, 当前线程数量:" + count.get());
return null;
}
}
// 如果队列不空，出队；如果队列已空且已经超时，返回null
Runnable task = queue.poll(keepAliveTime, TimeUnit.MILLISECONDS);
if (task == null) {
timeOut = true;
continue;
} else {
System.out.println(LocalTime.now() + ": " + threadName + " [出队] queue.poll()的超时方法得到task");
return task;
}
} else {
System.out.println(LocalTime.now() + ": " + threadName + " queue.take() 若队列空了，则一直阻塞，若不为空，则[出队]");
if (curThreadSize == corePoolSize) {
System.out.println(LocalTime.now() + ": " + threadName + " 当前为阻塞状态，线程池只能保留核心线程不会销毁,corePoolSize="+ corePoolSize + "(因阻塞状态故不会被销毁)，线程池继续等待新任务...\n");
}
// 若队列空了，则一直阻塞，若不为空，则出队
return queue.take();
}
}
}
}
调用执行
Demo 分2次进行测试线程池，主要是测试线程池能否被复用线程。
package com.sufadi.study.thread;
import java.time.LocalTime;
public class MyThreadPoolTest {
public static void main(String[] args) {
final MyThreadPool threadPool = new MyThreadPool();
new Thread(new Runnable() {
@Override
public void run() {
System.out.println(LocalTime.now() + ": 第一波：调用5次线程池，验证线程池如何实现线程的重复利用，原理为生产者和消费者模型");
for (int i = 0; i < 5; i++) {
threadPool.execute(new Runnable() {
@Override
public void run() {
System.out.println(LocalTime.now() + ": " + Thread.currentThread().getName() + ",[消费中]线程运行状态:" + Thread.currentThread().getState());
try {
Thread.sleep(500);
} catch (Exception e) {
e.printStackTrace();
}
System.out.println(LocalTime.now() + ": " + Thread.currentThread().getName() + ",[消费结束]");
}
});
}
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
System.out.println(LocalTime.now() + ": 延迟10秒，进行第二波线程池测试");
try {
Thread.sleep(10000);
} catch (Exception e) {
e.printStackTrace();
}
System.out.println(LocalTime.now() + ": 第二波：调用5次线程池，验证线程池如何实现线程的重复利用，原理为生产者和消费者模型");
for (int i = 0; i < 5; i++) {
threadPool.execute(new Runnable() {
@Override
public void run() {
System.out.println(LocalTime.now() + ": " + Thread.currentThread().getName() + ",[消费中]线程运行状态:" + Thread.currentThread().getState());
try {
Thread.sleep(2000);
} catch (Exception e) {
e.printStackTrace();
}
System.out.println(LocalTime.now() + ": " + Thread.currentThread().getName() + ",[消费结束]");
}
});
}
}
}).start();
}
}
运行结果
17:42:19.509: MyThreadPool corePoolSize:1, maxNumPoolSize:3, queue.size:0, keepAliveTime(ms):100
17:42:19.512: 第一波：调用5次线程池，验证线程池如何实现线程的重复利用，原理为生产者和消费者模型
17:42:19.513: 延迟10秒，进行第二波线程池测试
17:42:19.513: 1. 若 corePoolSize 未满，则开启核心线程, 当前线程数量:1
17:42:19.513: Thread-2,开始消费线程
17:42:19.513: 2. 若 corePoolSize 已满且queueSize未满，则[入队] 当前线程数量:1
17:42:19.513: 2. 若 corePoolSize 已满且queueSize未满，则[入队] 当前线程数量:1
17:42:19.513: Thread-2,[消费中]线程运行状态:RUNNABLE
17:42:19.513: 2. 若 corePoolSize 已满且queueSize未满，则[入队] 当前线程数量:1
17:42:19.514: 3. 若 queueSize 已满且maxNumPoolSize未满，则开启非核心线程, 当前线程数量:2
17:42:19.514: Thread-3,开始消费线程
17:42:19.514: Thread-3,[消费中]线程运行状态:RUNNABLE
17:42:20.014: Thread-2,[消费结束]
17:42:20.014: Thread-3,[消费结束]
17:42:20.014: Thread-2 [出队] queue.poll()的超时方法得到task
17:42:20.014: Thread-3 [出队] queue.poll()的超时方法得到task
17:42:20.014: Thread-2,开始消费线程
17:42:20.014: Thread-3,开始消费线程
17:42:20.014: Thread-3,[消费中]线程运行状态:RUNNABLE
17:42:20.014: Thread-2,[消费中]线程运行状态:RUNNABLE
17:42:20.515: Thread-3,[消费结束]
17:42:20.515: Thread-2,[消费结束]
17:42:20.515: Thread-3 [出队] queue.poll()的超时方法得到task
17:42:20.515: Thread-3,开始消费线程
17:42:20.515: Thread-3,[消费中]线程运行状态:RUNNABLE
17:42:20.615: Thread-2 已超时, 可以销毁了, 当前线程数量:1
17:42:21.015: Thread-3,[消费结束]
17:42:21.015: Thread-3 queue.take() 若队列空了，则一直阻塞，若不为空，则[出队]
17:42:21.015: Thread-3 当前为阻塞状态，线程池只能保留核心线程不会销毁,corePoolSize=1(因阻塞状态故不会被销毁)，线程池继续等待新任务...
17:42:29.513: 第二波：调用5次线程池，验证线程池如何实现线程的重复利用，原理为生产者和消费者模型
17:42:29.514: 2. 若 corePoolSize 已满且queueSize未满，则[入队] 当前线程数量:1
17:42:29.514: Thread-3,开始消费线程
17:42:29.514: 2. 若 corePoolSize 已满且queueSize未满，则[入队] 当前线程数量:1
17:42:29.514: 2. 若 corePoolSize 已满且queueSize未满，则[入队] 当前线程数量:1
17:42:29.514: Thread-3,[消费中]线程运行状态:RUNNABLE
17:42:29.514: 2. 若 corePoolSize 已满且queueSize未满，则[入队] 当前线程数量:1
17:42:29.514: 3. 若 queueSize 已满且maxNumPoolSize未满，则开启非核心线程, 当前线程数量:2
17:42:29.514: Thread-4,开始消费线程
17:42:29.514: Thread-4,[消费中]线程运行状态:RUNNABLE
17:42:31.514: Thread-3,[消费结束]
17:42:31.514: Thread-3 [出队] queue.poll()的超时方法得到task
17:42:31.514: Thread-3,开始消费线程
17:42:31.514: Thread-3,[消费中]线程运行状态:RUNNABLE
17:42:31.514: Thread-4,[消费结束]
17:42:31.514: Thread-4 [出队] queue.poll()的超时方法得到task
17:42:31.515: Thread-4,开始消费线程
17:42:31.515: Thread-4,[消费中]线程运行状态:RUNNABLE
17:42:33.514: Thread-3,[消费结束]
17:42:33.514: Thread-3 [出队] queue.poll()的超时方法得到task
17:42:33.515: Thread-3,开始消费线程
17:42:33.515: Thread-3,[消费中]线程运行状态:RUNNABLE
17:42:33.515: Thread-4,[消费结束]
17:42:33.615: Thread-4 已超时, 可以销毁了, 当前线程数量:1
17:42:35.515: Thread-3,[消费结束]
17:42:35.515: Thread-3 queue.take() 若队列空了，则一直阻塞，若不为空，则[出队]
17:42:35.515: Thread-3 当前为阻塞状态，线程池只能保留核心线程不会销毁,corePoolSize=1(因阻塞状态故不会被销毁)，线程池继续等待新任务...
|