手写线程池（复习自用）
- 介绍
- 代码
介绍
参考黑马满老师的JUC课程，给代码加上了相应的注释
如图所示，线程池核心线程数为1，任务队列的容量为1，假设要执行的线程数为4个，首先取一个放入线程池运行，第二个线程就要放入任务队列等待，第三个线程由于超过了任务队列的容量，就会执行拒绝策略。
代码
package com.test;
import lombok.extern.slf4j.Slf4j;
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
@Slf4j
public class TestPool {
public static void main(String[] args) {
// 创建线程池，依次输入线程池的核心线程数、超时时间、超时单位、任务队列的容量、拒绝策略（函数式接口）
ThreadPool threadPool = new ThreadPool(1,
1000, TimeUnit.MILLISECONDS, 1, ((queue, task) -> {
// 拒绝策略1：死等
// queue.put(task);
// 策略2：带超时等待
// queue.offer(task,500,TimeUnit.MILLISECONDS);
// 策略3：让调用者放弃任务执行
// log.info("放弃 {}",task);
// 策略4：让调用者抛出异常
// throw new RuntimeException("任务执行失败 "+task);
// 策略5：让调用者自己执行任务（也就是main线程）
task.run();
}));
// 模拟线程池执行4个任务（也就是有4个worker（worker继承自Thread类，内含Runnable成员变量）要线程池执行）
// 由于4个线程超过线程池的核心线程数，就会放入任务队列（也就是阻塞队列）进行等待，但由于任务队列的容量为1
// 也就是多余的线程暂时无法放入任务队列，就执行拒绝策略
for (int i = 0; i < 4; i++) {
// j是为了让i可以输出
int j = i;
threadPool.execute(()->{
try {
Thread.sleep(10000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
log.info("{}",j);
});
}
}
}
// 阻塞队列（用作任务队列，当线程池满了的话就任务放入任务队列）
@Slf4j
class BlockingQueue<T> {
// 1.任务队列
private Deque<T> queue = new ArrayDeque<>();
// 2.可重入锁
private ReentrantLock lock = new ReentrantLock();
// 下面两个条件变量的作用主要是用于往任务队列添加任务和取任务
// 防止空队列时取任务，满队列时加任务出错（类似操作系统中的PV操作）
// 3.生产者条件变量（往任务队列加内容）（使变满）
private Condition fullWaitSet = lock.newCondition();
// 4.消费者条件变量（从任务队列拿内容出来）（使变空）
private Condition emptyWaitSet = lock.newCondition();
// 5.容量
private int capacity;
public BlockingQueue(int capacity) {
this.capacity = capacity;
}
// 阻塞添加（往任务队列中添加线程）
public void put(T task) {
lock.lock();
try {
// 当任务队列达到容量时（往任务队列添加内容要先看看任务队列还有没有位置）
while (queue.size() == capacity) {
try {
// 也就是任务队列满了
log.info("等待加入任务队列{}...", task);
fullWaitSet.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// 当任务队列有空位置就加入任务
log.info("加入任务队列{}", task);
queue.addLast(task);
emptyWaitSet.signal();
} finally {
lock.unlock();
}
}
// 阻塞获取线程（也就是从任务队列中把任务提供给线程池）
public T take() {
lock.lock();
try {
// 当任务队列是空的时候就一直让消费者条件变量进行等待
// 因为从任务队列取内容要防止出现向空队列取任务的情况
while (queue.isEmpty()) {
try {
emptyWaitSet.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// 当任务队列不为空的时候，就将任务移除
// 唤醒生产者条件变量
T t = queue.removeFirst();
fullWaitSet.signal();
return t;
} finally {
lock.unlock();
}
}
// 下面是带有超时时间的阻塞添加和获取
// 阻塞添加
public boolean offer(T task, long timeout, TimeUnit timeUnit) {
lock.lock();
try {
// 先将超时时间转化为纳秒
long nanos = timeUnit.toNanos(timeout);
// 循环判断队满状态
while (queue.size() == capacity) {
try {
// 如果nacos（后面的代码让它在while循环中会不断减少）小于等于0
// 说明超时了，就直接返回false
if (nanos <= 0) {
return false;
}
log.info("{} 等待加入队列",task);
nanos = fullWaitSet.awaitNanos(nanos);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// 如果队列不满，就把任务加入队列
log.info("{} 加入队列",task);
queue.addLast(task);
emptyWaitSet.signal();
return true;
} finally {
lock.unlock();
}
}
// 阻塞获取
public T poll(long timeout,TimeUnit timeUnit){
lock.lock();
try{
// 计算剩余时间
long nanos = timeUnit.toNanos(timeout);
// 队列为空就一直循环
while (queue.isEmpty()) {
try {
if(nanos <= 0){
return null;
}
nanos = emptyWaitSet.awaitNanos(nanos);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// 队列不为空就取出任务
T t = queue.removeFirst();
// 唤醒生产者条件变量
fullWaitSet.signal();
return t;
}finally {
lock.unlock();
}
}
// 获取任务队列的长度
public int size(){
lock.lock();
try{
return queue.size();
}finally {
lock.unlock();
}
}
// 当任务队列满的时候可以选择拒绝策略
public void tryPut(RejectPolicy<T> rejectPolicy,T task){
lock.lock();
try {
// 当任务队列满的时候执行拒绝策略
if (queue.size() == capacity) {
rejectPolicy.reject(this,task);
}else{
log.info("{} 加入任务队列",task);
queue.addLast(task);
emptyWaitSet.signal();
}
}finally {
lock.unlock();
}
}
}
// 拒绝策略的接口，当队列满的时候任务队列可以选择拒绝策略
// 策略模式，这个拒绝策略可以自己写（函数式接口）
@FunctionalInterface
interface RejectPolicy<T>{
void reject(BlockingQueue<T> queue, T task);
}
@Slf4j
// 实现线程池
class ThreadPool{
// 任务队列
private BlockingQueue<Runnable> taskQueue;
// 线程集合（用hashset是为了去重）
// Worker是继承自Thread的内部类，下文可以看到
private HashSet<Worker> workers = new HashSet();
// 核心线程数
private int coreSize;
// 获取任务时的超时时间
private long timeout;
// 超时时间的单位
private TimeUnit timeUnit;
// 拒绝策略
private RejectPolicy<Runnable> rejectPolicy;
// 继承自Thread的内部类，里面有Runnable的成员变量、构造方法和run方法
class Worker extends Thread{
private Runnable task;
public Worker(Runnable task) {
this.task = task;
}
@Override
public void run() {
// 执行任务
// 1.当 task 不为空，执行任务
// 2.当 task 执行完毕，再接着从任务队列获取任务并执行
while(task != null || (task = taskQueue.poll(timeout,timeUnit))!= null){
try {
log.info("运行任务：",task);
task.run();
} catch (Exception e) {
e.printStackTrace();
}finally {
// 运行结束之后让任务为空，这样才可以继续从任务队列中取任务
task = null;
}
}
// 从线程集合中删除该Worker
synchronized (workers) {
log.info("worker 被移除{}",this);
workers.remove(this);
}
}
}
public ThreadPool( int coreSize, long timeout, TimeUnit timeUnit, int queueCapacity,RejectPolicy<Runnable> rejectPolicy) {
this.taskQueue = new BlockingQueue<>(queueCapacity);
this.coreSize = coreSize;
this.timeout = timeout;
this.timeUnit = timeUnit;
this.rejectPolicy = rejectPolicy;
}
// 线程池执行线程任务
public void execute(Runnable task){
// 当任务数没有超过coreSize时，直接交给worker对象（继承了Thread类）执行（start方法）
// 如果任务数超过了coreSize时，交给任务队列缓存（如果任务队列满了就执行拒绝策略）
synchronized (workers) {
if(workers.size() < coreSize){
Worker worker = new Worker(task);
log.info("新增 worker{} , {}",worker,task);
// 往线程池中加入(worker)线程
workers.add(worker);
// 运行线程
// 由Worker类中的run方法可以知道线程运行结束后会自动从线程池中移除
worker.start();
}else{
// 交给任务队列的拒绝策略
taskQueue.tryPut(rejectPolicy,task);
}
}
}
}
![[游戏开发][虚幻5]新建项目注意事项](https://img-blog.csdnimg.cn/direct/13ca23567b4844888aa638725ed63840.png)
