Java 自定义线程池实现

自定义线程池

简介
任务图示
阻塞队列 BlockingQueue<T>
- ReentrantLock
- 代码
线程池 ThreadPool
- 工作线程类 Worker
拒绝策略接口
代码测试类 TestThreadPool
- 为什么需要j = i？（lambad表达式相关）
测试结果
- 拒绝策略：让调用者自己执行任务
- 拒绝策略：让调用者放弃任务执行

简介

学习黑马JUC 并发编程过程中根据教程完成的线程池案例

任务图示

在这里插入图片描述
线程池Thread Pool保存工作资源，即n个工作线程，通过从任务队列Blocking Queue中获取m个任务执行。

阻塞队列 BlockingQueue

注意ReentrantLock类的使用

包括任务队列、生产者消费者锁、容量capcity、阻塞获取、带时阻塞获取、阻塞添加、带时阻塞添加、带拒绝策略的添加

由Blocking Queue对获取任务或者添加任务的操作进行阻塞管理，分别设置生产者锁（任务队列满则阻塞生产者，停止添加任务）、消费者锁（任务队列空则阻塞消费者，停止获取任务）。

ReentrantLock

基本语法：

reentrantLock.lock();
try{// 临界区
} finally {// 释放锁reentrantLock.unlock();
}

创建条件变量，以及阻塞和唤醒的基本使用
在这里插入图片描述

代码

@Slf4j(topic = "c.BlockingQueue")
class BlockingQueue<T> {// 1. 任务队列private Deque<T> queue = new ArrayDeque<>();// 2. 锁private ReentrantLock lock = new ReentrantLock();// 3. 生产者条件变量private Condition fullWaitSet = lock.newCondition();// 4. 消费者条件变量private Condition emptyWaitSet = lock.newCondition();// 5. 容量private int capcity;public BlockingQueue(int capcity) {this.capcity = capcity;}// 带超时阻塞获取public T poll(long timeout, TimeUnit unit) {lock.lock();try {//将timeout统一转换为 纳秒long nanos = unit.toNanos(timeout);//如果任务队列为空，在消费者阻塞队列中，超时时间内循环阻塞while(queue.isEmpty()){try{if (nanos <= 0){return null;}//超时时间内进行 空队列阻塞nanos = emptyWaitSet.awaitNanos(nanos);} catch (InterruptedException e) {throw new RuntimeException(e);}}T t = queue.removeFirst();//唤醒 生产者阻塞队列 中的线程fullWaitSet.signal();return t;}finally{lock.unlock();}}// 阻塞获取public T take() {lock.lock();try{//循环阻塞while(queue.isEmpty()){try {emptyWaitSet.await();}catch (InterruptedException e) {throw new RuntimeException(e);}}T t = queue.removeFirst();fullWaitSet.signal();return t;}finally{lock.unlock();}}// 阻塞添加public void put(T task) {lock.lock();try{while(queue.size() >= capcity){try{log.debug("等待加入任务队列————任务：{} ", task);fullWaitSet.await();} catch (InterruptedException e) {throw new RuntimeException(e);}}log.debug("加入任务队列————任务： {}", task);queue.addLast(task);emptyWaitSet.signal();}finally{lock.unlock();}}// 带超时时间阻塞添加public boolean offer(T task, long timeout, TimeUnit timeUnit) {lock.lock();try{long nanos = timeUnit.toNanos(timeout);while(queue.size() >= capcity){if(nanos <= 0) {return false;}try {log.debug("等待加入任务队列————任务：{} ", task);//awaitNanos返回剩余等待时间nanos = fullWaitSet.awaitNanos(nanos);} catch (InterruptedException e) {throw new RuntimeException(e);}}log.debug("加入任务队列————任务： {}", task);queue.addLast(task);emptyWaitSet.signal();return true;}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() >= capcity){//使用拒绝策略rejectPolicy.reject(this, task);} else{log.debug("加入任务队列————任务： {}", task);queue.addLast(task);emptyWaitSet.signal();}}finally{lock.unlock();}}
}

线程池 ThreadPool

包含阻塞队列、工作线程集合、线程数量、超时时间设置、执行任务

//线程池
@Slf4j(topic = "c.ThreadPool")
class ThreadPool {// 任务队列private BlockingQueue<Runnable> taskQueue;// 线程集合private HashSet<Worker> workers = new HashSet<>();// 核心线程数private int coreSize;// 获取任务时的超时时间private long timeout;private TimeUnit timeUnit;private RejectPolicy<Runnable> rejectPolicy;//初始化public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapcity,RejectPolicy<Runnable> rejectPolicy) {this.coreSize = coreSize;this.timeout = timeout;this.timeUnit = timeUnit;this.taskQueue = new BlockingQueue<>(queueCapcity);this.rejectPolicy = rejectPolicy;}// 执行任务public void execute(Runnable task) {synchronized (workers){if(workers.size() < coreSize){Worker worker = new Worker(task);log.debug("新增 worker 投入工作————工作线程：{}, 任务：{}", worker, task);workers.add(worker);worker.start();}else{taskQueue.tryPut(rejectPolicy, task);}}}class Worker extends Thread{...}
}

工作线程类 Worker

线程池ThreadPool的内部类，主要设置了线程的执行方法，
没有设置超时时间，线程执行完所有任务之后不会被销毁，而是继续等待更多任务，设置超时时间线程则会在长时间获取不到任务后结束执行
while(task != null || (task = taskQueue.take()) != null)循环获取任务队列中的任务并执行
while(task != null || (task = taskQueue.poll(timeout, timeUnit)) != null) 为带超时的执行方法，循环到超时时间则停止循环

一个线程与一个任务（不带超时）
一个线程与一个任务（带超时）

class Worker extends Thread{private Runnable task;public Worker(Runnable task){this.task = task;}@Overridepublic void run() {//task不为空执行任务，执行完毕再从任务队列中获取任务并执行//while(task != null || (task = taskQueue.take()) != null){//超时设置while(task != null || (task = taskQueue.poll(timeout, timeUnit)) != null){try {log.debug("线程正在执行————任务：{}", task);task.run();} catch(Exception e){e.printStackTrace();} finally {task = null;}}synchronized (workers){log.debug("worker 被移除————工作线程：{}", this);workers.remove(this);}}}

拒绝策略接口

在任务队列满时设置不同的策略处理之后再申请添加的队列

@FunctionalInterface // 拒绝策略
interface RejectPolicy<T> {void reject(BlockingQueue<T> queue, T task);
}

代码测试类 TestThreadPool

初始化ThreadPool时使用lambda表达式实现拒绝策略，threadPool.execute(() -> { } 传入的就是任务的内容，设置Thread.sleep(1000L) 用于测试拒绝策略

@Slf4j(topic = "c.TestThreadPool")
public class TestThreadPool{public static void main(String[] args) {ThreadPool threadPool = new ThreadPool(2,1000, TimeUnit.MILLISECONDS, 2, (queue, task)->{// 1. 死等// queue.put(task);// 2) 带超时等待// queue.offer(task, 1500, TimeUnit.MILLISECONDS);// 3) 让调用者放弃任务执行// log.debug("放弃{}", task);// 4) 让调用者抛出异常// throw new RuntimeException("任务执行 失败 " + task);// 5) 让调用者自己执行任务task.run();});for (int i = 0; i < 10; i++) {int j = i;threadPool.execute(() -> {try {Thread.sleep(1000L);} catch (InterruptedException e) {e.printStackTrace();}log.debug("{}", j);});}}
}

为什么需要j = i？（lambad表达式相关）

Lambda表达式内部引用的变量必须是final或者是事实上的final（即在Lambda表达式内部没有改变过它的值）。在Java 8之前，匿名内部类内部引用的局部变量必须声明为final。在Java 8中，这一规则放宽了一些，但要求这些变量的值在编译时确定，所以被称为"effectively final"（事实上的final）。

i虽然在每次循环中都有不同的值，但它是在Lambda表达式内部引用的。由于Lambda表达式的执行是延迟的，当Lambda表达式被执行时，循环可能已经结束，i的值可能已经改变。为了避免这种情况，需要将i赋值给一个final或者事实上的final变量j，以确保Lambda表达式内部引用的变量是不可变的。

因此，通过将i赋值给j，保证了在Lambda表达式内部引用的j是不可变的，从而避免了可能出现的并发问题。