自定义线程池：
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(topic = "c.DefineThreadPool")
public class DefineThreadPool {public static void main(String[] args) {ThreadPool threadPool = new ThreadPool(1,100,TimeUnit.MILLISECONDS,1,(queue,task)->{//1.死等// queue.put(task);//2.带超时等待// queue.offer(task,1500,TimeUnit.MILLISECONDS);//3.让调用者放弃任务执行// System.out.println("放弃任务..."+task);//4.让调用者抛出异常//throw new RuntimeException("任务执行失败 "+task);//5.让调用者自己执行任务//task.run();});for (int i = 0; i < 3;i++){int j = i;threadPool.execute(() -> {try {Thread.sleep(1000L);} catch (InterruptedException e) {e.printStackTrace();}log.debug("{}", j);});}}}@FunctionalInterface //拒绝策略
interface RejectPolicy<T>{void reject(BlockingQueue<T> queue,T task);
}@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 void execute(Runnable task){//当任务数没有超过 coreSize 时，直接交给worker 对象执行//如果任务数超过coreSize 时，加入任务队列暂存if (workers.size() < coreSize){Worker worker = new Worker(task);System.out.println("新增 worker"+worker+","+task.toString());workers.add(worker);worker.start();}else {// taskQueue.put(task);//拒绝策略类型，常见的几种// 1）死等// 2）带超时等待// 3）让调用者放弃任务执行// 4) 让调用者抛出异常// 5）让调用者自己执行任务taskQueue.tryPut(rejectPolicy,task);}}public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit,int queueCapacity,RejectPolicy<Runnable> rejectPolicy) {this.coreSize = coreSize;this.timeout = timeout;this.timeUnit = timeUnit;this.taskQueue = new BlockingQueue<>(queueCapacity);this.rejectPolicy = rejectPolicy;}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){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("移除了，，，"+this);workers.remove(this);}}}
}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 capacity;public BlockingQueue(int capacity) {this.capacity = capacity;}//带超时的阻塞获取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){e.printStackTrace();}}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){e.printStackTrace();}}T t = queue.removeFirst();fullWaitSet.signal();return t;}finally {lock.unlock();}}//阻塞添加public void put(T task){lock.lock();try {while (queue.size() == capacity){try {System.out.println("等待加入任务队列"+task);fullWaitSet.await();}catch (InterruptedException e){e.printStackTrace();}}System.out.println("加入任务队列"+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() == 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<T> 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();}}
}