实现java简单的线程池实现java简单的线程池

拆分实现流程
实现方式

1.拒绝策略
2.阻塞队列
3.线程池和工作线程
策略模式

对比JDK的线程池

线程池的状态转化

总结

拆分实现流程请看下面这张图

文章图片

首先我们得对线程池进行一个功能拆分

Thread Pool 就是我们的线程池，t1，t2，t3代表三个线程
Blocking Queue代表阻塞队列
main代表main方法的线程
task1，task2，task3代表要执行的每个任务

现在我们梳理一下执行的流程，注意这里是简略版的，文章后面我会给出详细版的

文章图片

所以此时，我们发现了需要创建几个类，或者说几个角色，分别是

线程池
工作线程
阻塞队列
拒绝策略（干嘛的？就是当线程数已经满了，并且阻塞队列也满了，还有任务想进入阻塞队列的时候，就可以拒绝这个任务）

实现方式
1.拒绝策略

/** * 拒绝策略 */@FunctionalInterfaceinterface RejectPolicy{ //queue就是我们自己实现的阻塞队列，task是任务void reject(BlockingQueue queue,T task); }

2.阻塞队列
我们需要实现四个方法，获取和添加，超时获取和超时添加，至于方法实现的细节，我都备注了大量的注释进行解释。

/** * 阻塞队列 */class BlockingQueue{//阻塞队列private Deque queue = new ArrayDeque<>(); //锁private ReentrantLock lock = new ReentrantLock(); //生产者条件变量private Condition fullWaitSet = lock.newCondition(); //消费者条件变量private Condition emptyWaitSet = lock.newCondition(); //容量private int capacity; public BlockingQueue(int capacity){this.capacity = capacity; }//带有超时阻塞获取public T poll(long timeout, TimeUnit timeUnit){lock.lock(); try {//将timeout统一转换为纳秒long nanos = timeUnit.toNanos(timeout); while(queue.isEmpty()){try {if(nanos <= 0){//小于0，说明上次没有获取到，代表已经超时了return null; }//返回值是剩余的时间nanos = emptyWaitSet.awaitNanos(nanos); } catch (InterruptedException e) {e.printStackTrace(); }}T t = queue.removeFirst(); //通知生产者fullWaitSet.signal(); return t; }finally {lock.unlock(); }}//阻塞获取public T take(){lock.lock(); try{while(queue.isEmpty()){ //如果任务队列为空，代表线程池没有可以执行的内容try {/*也就说此时进来的线程是执行不了任务的，所以此时emptyWaitSet消费者要进行阻塞状态等待下一次唤醒，然后继续判断队列是否为空*/emptyWaitSet.await(); } catch (InterruptedException e) {e.printStackTrace(); }}/*代码执行到这里。说明任务队列不为空，线程池就从任务队列拿出一个任务出来执行也就是说把阻塞队列的一个任务出队*/T t = queue.removeFirst(); /*然后唤醒之前存放在生成者Condition休息室，因为由于之前阻塞队列已满，fullWaitSet才会进入阻塞状态所以当阻塞队列删除了任务，就要唤醒之前进入阻塞状态的fullWaitSet*/fullWaitSet.signal(); //返回任务return t; }finally {lock.unlock(); }}//阻塞添加public void put(T task){lock.lock(); try {while(queue.size() == capacity){//任务队列满了try {System.out.println("等待加入任务队列"+task); /*也就说此时进来的任务是进不了阻塞队列的，已经满了，所以此时生产者Condition要进入阻塞状态等待下一次唤醒，然后继续判断队列是否为空*/fullWaitSet.await(); } catch (InterruptedException e) {e.printStackTrace(); }}//任务队列还未满System.out.println("加入任务队列"+task); //把任务加入阻塞队列queue.addLast(task); /*然后唤醒之前存放在消费者Condition休息室，因为由于之前阻塞队列为空，emptyWaitSet才会进入阻塞状态所以当阻塞队列加入了任务，就要唤醒之前进入阻塞状态的emptyWaitSet*/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() == capacity){try {if(nanos < 0){return false; }System.out.println("等待加入任务队列"+task); //不会一直阻塞，超时就会继续向下执行nanos = fullWaitSet.awaitNanos(nanos); } catch (InterruptedException e) {e.printStackTrace(); }}System.out.println("加入任务队列"+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 rejectPolicy, T task){lock.lock(); try {//判断队列是否已满if(queue.size() == capacity){rejectPolicy.reject(this,task); }else{//有空闲System.out.println("加入任务队列"+task); queue.addLast(task); emptyWaitSet.signal(); }}finally {lock.unlock(); }}}

3.线程池和工作线程
我把工作线程当成线程池的内部类去实现。方便调用变量。

/** * 线程池 */class ThreadPool{//阻塞队列private BlockingQueue taskQueue; //线程集合private HashSet workers = new HashSet<>(); //核心线程数private int coreSize; //获取任务的超时时间private long timeout; private TimeUnit timeUnit; private RejectPolicy rejectPolicy; public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapacity,RejectPolicy rejectPolicy) {this.coreSize = coreSize; this.timeout = timeout; this.timeUnit = timeUnit; this.taskQueue = new BlockingQueue<>(queueCapacity); this.rejectPolicy = rejectPolicy; }//执行任务public void execute(Runnable task){synchronized (workers){if(workers.size() <= coreSize){//当前的线程数小于核心线程数Worker worker = new Worker(task); workers.add(worker); //让线程开始工作，执行它的run方法worker.start(); }else{// 1) 死等// 2) 带超时等待// 3) 让调用者放弃任务执行// 4) 让调用者抛出异常// 5) 让调用者自己执行任务taskQueue.tryPut(rejectPolicy,task); }}}/*** 工作线程，也就是线程池里面的线程*/class Worker extends Thread{private Runnable task; public Worker(Runnable task){this.task = task; }@Overridepublic void run() {//执行任务// 1) 当 task 不为空，执行任务// 2) 当 task 执行完毕，再接着从任务队列获取任务并执行while (task != null || (task = taskQueue.poll(timeout, timeUnit)) != null) {try {System.out.println("正在执行的任务" + task); task.run(); } catch (Exception e) {e.printStackTrace(); } finally {//代表这个任务已经执行完了task = null; }}synchronized (workers) {System.out.println("worker 被移除" + this); workers.remove(this); }}}}

策略模式
细心的小伙伴已经发现，我在拒绝策略这里使用了23种设计模式的策略模式，因为我没有将拒绝的方式写死，而是交给了调用者去实现。

对比JDK的线程池下面是JDK自带的线程池