自定义线程池

步骤1：自定义阻塞队列

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) {
 　　　　　　　　　　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() == capcity) {
 　　　　　　　　try {
 　　　　　　　　　　log.debug("等待加入任务队列 {} ...", task);
 　　　　　　　　　　fullWaitSet.await();
 　　　　　　　　} catch (InterruptedException e) {
 　　　　　　　　　　e.printStackTrace();
 　　　　　　　　}
 　　　　　　}
 　　　　　　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) {
 　　　　　　　　try {
 　　　　　　　　　　if(nanos <= 0) {
 　　　　　　　　　　　　return false;
 　　　　　　　　　　}
 　　　　　　　　　　log.debug("等待加入任务队列 {} ...", task);
 　　　　　　　　　　nanos = fullWaitSet.awaitNanos(nanos);
　　　　　　　　} catch (InterruptedException e) {
 　　　　　　　　　　e.printStackTrace();
 　　　　　　　　}
 　　　　　　}
 　　　　　　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();
 　　　　}
　　 }
}

步骤2：自定义线程池

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 对象，并交给 worker 对象执行
 　　　　// 如果任务数超过 coreSize 时，不会再新建 worker 对象。加入任务队列暂存，等待已有的 worker 线程执行。
 　　　　synchronized (workers) {
 　　　　　　if(workers.size() < coreSize) {
 　　　　　　　　Worker worker = new Worker(task);
 　　　　　　　　log.debug("新增 worker{}, {}", worker, task);
 　　　　　　　　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 queueCapcity, 
　　　　 RejectPolicy<Runnable> rejectPolicy) {
 　　　　this.coreSize = coreSize;
 　　　　this.timeout = timeout;
 　　　　this.timeUnit = timeUnit;
 　　　　this.taskQueue = new BlockingQueue<>(queueCapcity);
 　　　　　　this.rejectPolicy = rejectPolicy;
 　　　　}
 　　　　
　　　　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.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);
 　　　　　　　　}
 　　　　　　}
 　　　　}
}

测试：

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, 1500, TimeUnit.MILLISECONDS);
 　　　　// 3) 让调用者放弃任务执行
　　　　// log.debug("放弃{}", task);
 　　　　// 4) 让调用者抛出异常
　　　　// throw new RuntimeException("任务执行失败 " + task);
 　　　　// 5) 让调用者自己执行任务
 　　　　task.run();
 　　});
 　　for (int i = 0; i < 4; i++) {
 　　　　int j = i;
 　　　　threadPool.execute(() -> {
　　　　　　 try {
 　　　　　　　　Thread.sleep(1000L);
 　　　　　　} catch (InterruptedException e) {
 　　　　　　　　e.printStackTrace();
 　　　　　　}
 　　　　　　log.debug("{}", j);
 　　　　});
 　　}
 }

不带超时的 take

即时队列空， worker 线程也会一直阻塞等待。也就是说不会走到下面从 workers（一个set） 里面移除当前 worker

带超时的 take

队列空，阻塞等待一段时间就结束了。就会走到下面的逻辑，从 workers(一个set) 里移除当前 worker