43.自定义线程池（一）

ThreadPool是线程池，里面是一定数量的线程，是消费者。

BlockingQueue阻塞队列，线程池中的线程会从阻塞队列中去拿任务执行。任务多了线程池处理不过来了，就会到Blocking Queue中排队，等待执行。链表结构，特点是先进先出。java中Deque是一个双向链表，操作起来更方便。

main就是生产者，不断产生新的执行任务。

package com.xkj.thread.pool;import java.util.ArrayDeque;
import java.util.Deque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;public class BlockingQueue<T> {//1.任务队列private Deque<T> queue = new ArrayDeque<>();//2.锁private Lock 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;}/*** 带超时的获取元素* @param timeout* @param unit* @return*/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;}//阻塞等待，当被唤醒后，队列不会空，不满足while条件，程序继续向下执行//返回的是timeout - 已经等待的时间 = 剩余的时间//防止虚假唤醒nanos = emptyWaitSet.awaitNanos(nanos);} catch (InterruptedException e) {e.printStackTrace();}}//获取队列头部的元素返回,获取元素后应该从队列中移除T t = queue.removeFirst();//唤醒生产者，继续添加元素fullWaitSet.signal();return t;}finally {lock.unlock();}}/*** 获取元素* @return*/public T take() {lock.lock();try {while (queue.isEmpty()) { //判断队列是否为空try {//阻塞等待，当被唤醒后，队列不会空，不满足while条件，程序继续向下执行emptyWaitSet.await();} catch (InterruptedException e) {e.printStackTrace();}}//获取队列头部的元素返回,获取元素后应该从队列中移除T t = queue.removeFirst();//唤醒生产者，继续添加元素fullWaitSet.signal();return t;}finally {lock.unlock();}}/*** 添加元素* @param element*/public void put(T element) {lock.lock();try {while (queue.size() == capcity){try {fullWaitSet.await();} catch (InterruptedException e) {e.printStackTrace();}}queue.addLast(element);//唤醒消费者，继续获取任务emptyWaitSet.signal();}finally {lock.unlock();}}/*** 获取大小* @return*/public int size() {lock.lock();try {return queue.size();}finally {lock.unlock();}}
}

package com.xkj.thread.pool;import lombok.extern.slf4j.Slf4j;import java.util.HashSet;
import java.util.concurrent.TimeUnit;@Slf4j(topic = "c.ThreadPool")
public class ThreadPool {//任务队列private BlockingQueue<Runnable> taskQueue;//线程集合private HashSet<Worker> workers = new HashSet<>();//核心线程数private int coreSize;//获取任务的超时时间private long timeout;private TimeUnit timeUnit;public ThreadPool(int coreSize, int queueCapcity,long timeout, TimeUnit timeUnit) {this.coreSize = coreSize;this.timeout = timeout;this.timeUnit = timeUnit;this.taskQueue = new BlockingQueue<>(queueCapcity);}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.take()) != null) {try {log.debug("正在执行...{}", task);task.run();}catch (Exception e) {}finally {task = null;}}synchronized (workers) {log.debug("worker 被移除{}", this);workers.remove(this);}}}//执行任务public void execute(Runnable task) {synchronized (workers) {if(workers.size() < coreSize) {Worker worker = new Worker(task);log.debug("新增worker{},{}", worker, task);// 当任务数没有超过coreSize时，直接交给worker对象执行workers.add(worker);worker.start();} else {// 当任务数超过coreSize时，加入任务队列暂存log.debug("加入任务队列{}", task);taskQueue.put(task);}}}
}

@Slf4j(topic = "c.TestPool")
public class TestPool {public static void main(String[] args) {ThreadPool threadPool = new ThreadPool(2, 10, 1000, TimeUnit.MILLISECONDS);for (int i = 0; i < 5; i++) {int j = i;threadPool.execute(() -> {log.debug("{}", j);});}}
}