当有任务出现，如果每个任务均需创建一个新线程去处理任务，会过度浪费系统资源。线程池的做法则是：将线程存储起来，重复利用线程去处理任务。

image.png
整体是生产者-消费者结构，左侧Thread Pool存放可重用线程，中间的Blocking Queue用来平衡速度差异。Thread Pool消费tasks，main生产tasks.
如果tasks过多，线程处理不过来，Blocking Queue用来存储tasks，如果tasks过少，线程池中的消费者线程则需要在Blocking Queue中等待.

阻塞队列实现：

  2. class BlockingQueue<T>{
  4.     // 1.任务队列
  5.     private Deque<T> queue = new ArrayDeque<>();
  7.     // 2.锁
  8.     private ReentrantLock lock = new ReentrantLock();
  10.     // 3.生产者条件变量
  11.     private Condition fuulWaitSet = lock.newCondition();
  13.     // 4.消费者条件变量
  14.     private Condition emptyWaitSet = lock.newCondition();
  16.     // 5.容量
  17.     private int capcity;
  19.     public BlockingQueue(int capcity) {
  20.         this.capcity = capcity;
  21.     }
  23.     //带超时的阻塞获取
  24.     public T poll(long timeout, TimeUnit unit){
  25.         lock.lock();
  26.         try {
  27.             // 将timeout 统一转换为纳秒
  28.             long nanos = unit.toNanos(timeout);
  29.             while(queue.isEmpty()){
  30.                 try {
  31.                     if(nanos<=0){
  32.                         return null;
  33.                     }
  34.                     emptyWaitSet.awaitNanos(nanos);//能够自动解决虚假唤醒问题，返回的是剩余时间
  35.                 } catch (InterruptedException e) {
  36.                     e.printStackTrace();
  37.                 }
  38.             }
  39.             //获取队列头元素并返回
  40.             T t =queue.removeFirst();
  41.             fuulWaitSet.signal();
  42.             return t;
  43.         }finally {
  44.             lock.unlock();
  45.         }
  46.     }
  48.     // 阻塞获取
  49.     public T take(){
  50.         lock.lock();
  51.         try {
  52.             while(queue.isEmpty()){
  53.                 try {
  54.                     emptyWaitSet.await();
  55.                 } catch (InterruptedException e) {
  56.                     e.printStackTrace();
  57.                 }
  58.             }
  59.             //获取队列头元素并返回
  60.             T t =queue.removeFirst();
  61.             fuulWaitSet.signal();
  62.             return t;
  63.         }finally {
  64.             lock.unlock();
  65.         }
  66.     }
  68.     //阻塞添加
  69.     public void put(T element){
  70.         lock.lock();
  71.         try{
  72.             while(queue.size()==capcity){
  73.                 try {
  74.                     fuulWaitSet.await();
  75.                 } catch (InterruptedException e) {
  76.                     e.printStackTrace();
  77.                 }
  78.             }
  79.             queue.addLast(element);
  80.             emptyWaitSet.signal();
  81.         }finally {
  82.             lock.unlock();
  83.         }
  84.     }
  86.     //获取大小,当然这里可以不加锁，完全只读
  87.     public int size(){
  88.         lock.lock();
  89.         try {
  90.             return queue.size();
  91.         }finally {
  92.             lock.unlock();
  93.         }
  94.     }
  95. }

阻塞队列的实现：ReentrantLock锁+两种情况的wait-notify实现（队列空、队列满）

线程池实现(无超时时间)：

  1. class ThreadPool{
  3.     //线程池中需要用到阻塞队列(任务队列)
  4.     private BlockingQueue<Runnable> taskQueue;
  6.     // 线程集合,如果是Thread对象所包含的信息有限，所以将线程类包装成Worker类
  7.     private HashSet<Worker> workers = new HashSet<>();
  9.     // 核心线程数
  10.     private int coreSize;
  12.     // 获取任务的超时时间,如果线程限定时间内获取不到任务，释放线程
  13.     private long timeout;
  15.     private TimeUnit timeUnit;
  17.     // 执行任务
  18.     public void execute(Runnable task){
  19.         //当任务数没有超过 coreSize时，直接交给 worker对象执行
  20.         //如果任务数超过 coreSize时，加入任务队列暂存。
  21.         synchronized (workers){
  22.             //线程数小于任务
  23.             if (workers.size() < coreSize){
  24.                 Worker worker = new Worker(task);
  25.                 workers.add(worker);
  26.                 worker.start();
  27.             }else {
  28.                 taskQueue.put(task);
  29.             }
  30.         }
  31.     }
  34.     public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit,int queueCapacity) {
  35.         this.coreSize = coreSize;
  36.         this.timeout = timeout;
  37.         this.timeUnit = timeUnit;
  38.         this.taskQueue = new BlockingQueue<>(queueCapacity);
  39.     }
  42.     class Worker extends Thread{
  43.         private Runnable task;
  45.         public Worker(Runnable task){
  46.             this.task = task;
  47.         }
  49.         @Override
  50.         public void run() {
  51.             //执行任务
  52.             // 1 当 task不为空，执行任务
  53.             // 2 当task执行完毕，再接着从任务队列获取任务并执行
  55.             while(task!=null||(task = taskQueue.take())!=null)){
  56.                 try {
  57.                     task.run();
  58.                 }catch (Exception e){
  60.                 }
  61.                 finally {
  62.                     task = null;
  63.                 }
  64.             }
  65.             synchronized (workers){
  66.               workers.remove(this);
  67.             }
  68.         }
  69.     }
  70. }

本段代码逻辑稍微复杂些，ThreadPool中的 coreSize 表示线程池能承受的最大线程数。

这段代码实现的逻辑是：线程池中每个线程处理一个任务，传进task任务后，如果线程池当前线程数还没到满，则新创建线程并执行该任务，如果线程池已经满，则没有线程可以处理该任务，则将该任务存入任务队列中。

  1.     public void execute(Runnable task){
  2.         //当任务数没有超过 coreSize时，直接交给 worker对象执行
  3.         //如果任务数超过 coreSize时，加入任务队列暂存。
  5.         synchronized (workers){
  6.             //线程数小于任务
  7.             if (workers.size() < coreSize){
  8.                 Worker worker = new Worker(task);
  9.                 workers.add(worker);
  10.                 worker.start();
  11.             }else {
  12.                 taskQueue.put(task);
  13.             }
  14.         }
  15.     }

Worker implements Thread，所以Worker类是线程类，所以其要重写run方法，run方法要根据构造方法中传入的task来判断此任务是否可以执行。
注意run方法处于等待任务状态，如果新传进来的任务已经执行完毕，则其会去阻塞队列中查找任务执行，如果阻塞队列中已经没有任务可以执行，那么就会陷入【WAITING】状态，take（）的实现细节：如果队列为空，则一直等待下去。
这种思路的弊端是：线程如果没有任务可执行，会陷入空等，下边的销毁线程代码实际没有起到作用
注意taskQueue.task()方法的线程安全问题，已经在BlockingQueue类中的解决。

  1.         @Override
  2.         public void run() {
  3.             //执行任务
  4.             // 1 当 task不为空，执行任务
  5.             // 2 当task执行完毕，再接着从任务队列获取任务并执行
  7.             while(task!=null||(task = taskQueue.take())!=null)){
  8.                 try {
  9.                     task.run();
  10.                 }catch (Exception e){
  12.                 }
  13.                 finally {
  14.                     task = null;
  15.                 }
  16.             }
  17.             synchronized (workers){
  18.               workers.remove(this);
  19.             }
  20.         }


线程池实现(有超时时间)：

  2.     class Worker extends Thread{
  3.         private Runnable task;
  5.         public Worker(Runnable task){
  6.             this.task = task;
  7.         }
  9.         @Override
  10.         public void run() {
  11.             //执行任务
  12.             // 1 当 task不为空，执行任务
  13.             // 2 当task执行完毕，再接着从任务队列获取任务并执行
  14.             while(task!=null||(task = taskQueue.poll(timeout,timeUnit))!=null)){
  15.                 try {
  16.                     task.run();
  17.                 }catch (Exception e){
  19.                 }
  20.                 finally {
  21.                     task = null;
  22.                 }
  23.             }
  24.             synchronized (workers){
  25.               workers.remove(this);
  26.             }
  27.         }
  28.     }

Worker类中run方法实现细节改变，taskQueue.take()改为taskQueue.poll()，如果超时等待不到任务，taskQueue.poll()返回null值，run方法内的while循环结束，线程经过workers.remove()方法得以销毁。

阻塞队列满-拒绝策略：

如果任务数过多，阻塞队列中存放不下，主线程会陷入等待状态，应该使得主线程有选择空间。

  1. public class TestThreadPool {
  2.     public static void main(String[] args) {
  3.         //由主线程向线程池中提交任务
  4.         ThreadPool threadPool = new ThreadPool(2,1000,TimeUnit.MILLISECONDS,10);
  5.         for (int i=0;i<15;i++){
  6.             int j =i;
  7.             threadPool.execute(()->{
  8.                 System.out.println(j);
  9.             });
  10.         }
  11.     }
  12. }

向阻塞队列中输入任务的代码改为超时限制的添加：

  1.     //带超时时间的阻塞添加
  2.     public boolean offer(T task, long timeout, TimeUnit timeUnit){
  3.         lock.lock();
  4.         try {
  5.             long nanos = timeUnit.toNanos(timeout);
  6.             while (queue.size()==capcity){
  7.                 try {
  8.                     if(nanos<=0){
  9.                         return false;
  10.                     }
  11.                     nanos = fuulWaitSet.awaitNanos(nanos);
  12.                 } catch (InterruptedException e) {
  13.                     e.printStackTrace();
  14.                 }
  15.             }
  16.             queue.addLast(task);
  17.             emptyWaitSet.signal();
  18.             return true;
  19.         }finally {
  20.             lock.unlock();
  21.         }
  22.     }

向阻塞队列中输入任务有多种实现策略，如下：死等（最原始）、超时等待、调用者放弃任务的执行、调用者抛出异常、调用者自己执行任务

  1.     public void execute(Runnable task){
  2.         //当任务数没有超过 coreSize时，直接交给 worker对象执行
  3.         //如果任务数超过 coreSize时，加入任务队列缓存。
  5.         synchronized (workers){
  6.             //线程数小于任务
  7.             if (workers.size() < coreSize){
  8.                 Worker worker = new Worker(task);
  9.                 workers.add(worker);
  10.                 worker.start();
  11.             }else {
  12.                 taskQueue.put(task);
  13.                 // 1.死等
  14.                 // 2.带超时等待
  15.                 // 3.让调用者放弃任务执行
  16.                 // 4.让调用者抛出异常
  17.                 // 5.让调用者自己执行任务
  18.             }
  19.         }
  20.     }

如果将五种代码逻辑均写到execute方法中（多个if-else），明显代码灵活度不够，解决办法：策略模式，将方法抽象成接口，具体选择那种实现方式交给调用者，通过调用时传递进来。

声明函数式接口：

  1. @FunctionalInterface //拒绝策略，函数式接口
  2. interface RejectPolicy<T>{
  3.     void reject(BlockingQueue<T> queue,T task);
  4. }

BlockingQueue类中，增加tryPut方法：

  1.     public void tryPut(RejectPolicy<T> rejectPolicy, T task){
  2.         lock.lock();
  3.         try {
  4.             //判断队列是否为满
  5.             if(queue.size()==capcity){
  6.                 //传入就好，权力下放给调用者，妙！
  7.                 rejectPolicy.reject(this,task);
  8.             }else{
  9.                 queue.addLast(task);
  10.             }
  11.         }
  12.         finally {
  14.         }
  15.     }

如果队列满，那么调用rejectPolicy.reject()方法，将阻塞队列与task传进去，因为毕竟此操作是对阻塞队列而言。

因为tryPut方法中要使用rejectPolicy对象，且ThreadPool类中的execute方法需要调用tryPut方法，所以ThreadPool类中需要添加RejectPolicy类引用属性，并在构造方法中初始化：

  1.     public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit,int queueCapacity, RejectPolicy<Runnable> rejectPolicy) {
  2.         this.coreSize = coreSize;
  3.         this.timeout = timeout;
  4.         this.timeUnit = timeUnit;
  5.         this.taskQueue = new BlockingQueue<>(queueCapacity);
  6.         this.rejectPolicy = rejectPolicy;
  7.     }

主线程内，在ThreadPool初始化时，指定RejectPolicy的实现类对象：

  1.         ThreadPool threadPool = new ThreadPool(2,1000,TimeUnit.MILLISECONDS,10,(queue,task)->{
  2.             // 1.死等
  3.             queue.put(task);
  4.         });

本例完整代码：

  1. import com.sun.corba.se.spi.orbutil.threadpool.Work;
  2. import javafx.concurrent.Worker;
  4. import java.io.Closeable;
  5. import java.sql.Time;
  6. import java.util.ArrayDeque;
  7. import java.util.Deque;
  8. import java.util.HashSet;
  9. import java.util.concurrent.TimeUnit;
  10. import java.util.concurrent.locks.Condition;
  11. import java.util.concurrent.locks.ReentrantLock;
  13. public class TestThreadPool {
  14.     public static void main(String[] args) {
  15.         //由主线程向线程池中提交任务
  16.         ThreadPool threadPool = new ThreadPool(2,1000,TimeUnit.MILLISECONDS,10,(queue,task)->{
  17.             // 1.死等
  18.             queue.put(task);
  19.         });
  20.         for (int i=0;i<15;i++){
  21.             int j =i;
  22.             threadPool.execute(()->{
  23.                 System.out.println(j);
  24.             });
  25.         }
  26.     }
  27. }
  29. @FunctionalInterface //拒绝策略，函数式接口
  30. interface RejectPolicy<T>{
  31.     void reject(BlockingQueue<T> queue,T task);
  32. }
  35. class ThreadPool{
  36.     //线程池中需要用到阻塞队列,任务队列
  37.     private BlockingQueue<Runnable> taskQueue;
  39.     // 线程集合,如果是Thread对象所包含的信息有限，所以将线程类包装成Worker类
  40.     private HashSet<Worker> workers = new HashSet<>();
  42.     // 核心线程数
  43.     private int coreSize;
  45.     // 获取任务的超时时间,如果线程限定时间内获取不到任务，释放线程
  46.     private long timeout;
  48.     private TimeUnit timeUnit;
  50.     private RejectPolicy<Runnable> rejectPolicy;
  52.     // 执行任务
  53.     public void execute(Runnable task){
  54.         //当任务数没有超过 coreSize时，直接交给 worker对象执行
  55.         //如果任务数超过 coreSize时，加入任务队列缓存。
  57.         synchronized (workers){
  58.             //线程数小于任务
  59.             if (workers.size() < coreSize){
  60.                 Worker worker = new Worker(task);
  61.                 workers.add(worker);
  62.                 worker.start();
  63.             }else {
  64.                 taskQueue.put(task);
  65.                 // 1.死等
  66.                 // 2.带超时等待
  67.                 // 3.让调用者放弃任务执行
  68.                 // 4.让调用者抛出异常
  69.                 // 5.让调用者自己执行任务
  70.                 taskQueue.tryPut(rejectPolicy,task);
  71.             }
  72.         }
  73.     }
  76.     public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit,int queueCapacity, RejectPolicy<Runnable> rejectPolicy) {
  77.         this.coreSize = coreSize;
  78.         this.timeout = timeout;
  79.         this.timeUnit = timeUnit;
  80.         this.taskQueue = new BlockingQueue<>(queueCapacity);
  81.         this.rejectPolicy = rejectPolicy;
  82.     }
  84.     class Worker extends Thread{
  85.         private Runnable task;
  87.         public Worker(Runnable task){
  88.             this.task = task;
  89.         }
  91.         @Override
  92.         public void run() {
  93.             //执行任务
  94.             // 1 当 task不为空，执行任务
  95.             // 2 当task执行完毕，再接着从任务队列获取任务并执行
  96.             while(task!=null||(task = taskQueue.poll(timeout,timeUnit))!=null)){
  97.                 try {
  98.                     task.run();
  99.                 }catch (Exception e){
  101.                 }
  102.                 finally {
  103.                     task = null;
  104.                 }
  105.             }
  106.             synchronized (workers){
  107.               workers.remove(this);
  108.             }
  109.         }
  110.     }
  111. }
  114. class BlockingQueue<T>{
  116.     // 1.任务队列
  117.     private Deque<T> queue = new ArrayDeque<>();
  119.     // 2.锁
  120.     private ReentrantLock lock = new ReentrantLock();
  122.     // 3.生产者条件变量
  123.     private Condition fuulWaitSet = lock.newCondition();
  125.     // 4.消费者条件变量
  126.     private Condition emptyWaitSet = lock.newCondition();
  128.     // 5.容量
  129.     private int capcity;
  131.     public BlockingQueue(int capcity) {
  132.         this.capcity = capcity;
  133.     }
  135.     //带超时的阻塞获取
  136.     public T poll(long timeout, TimeUnit unit){
  137.         lock.lock();
  138.         try {
  139.             // 将timeout 统一转换为纳秒
  140.             long nanos = unit.toNanos(timeout);
  141.             while(queue.isEmpty()){
  142.                 try {
  143.                     if(nanos<=0){
  144.                         return null;
  145.                     }
  146.                     emptyWaitSet.awaitNanos(nanos);//能够自动解决虚假唤醒问题，返回的是剩余时间
  147.                 } catch (InterruptedException e) {
  148.                     e.printStackTrace();
  149.                 }
  150.             }
  151.             //获取队列头元素并返回
  152.             T t =queue.removeFirst();
  153.             fuulWaitSet.signal();
  154.             return t;
  155.         }finally {
  156.             lock.unlock();
  157.         }
  158.     }
  160.     // 阻塞获取
  161.     public T take(){
  162.         lock.lock();
  163.         try {
  164.             while(queue.isEmpty()){
  165.                 try {
  166.                     emptyWaitSet.await();
  167.                 } catch (InterruptedException e) {
  168.                     e.printStackTrace();
  169.                 }
  170.             }
  171.             //获取队列头元素并返回
  172.             T t =queue.removeFirst();
  173.             fuulWaitSet.signal();
  174.             return t;
  175.         }finally {
  176.             lock.unlock();
  177.         }
  178.     }
  180.     //阻塞添加
  181.     public void put(T element){
  182.         lock.lock();
  183.         try{
  184.             while(queue.size()==capcity){
  185.                 try {
  186.                     fuulWaitSet.await();
  187.                 } catch (InterruptedException e) {
  188.                     e.printStackTrace();
  189.                 }
  190.             }
  191.             queue.addLast(element);
  192.             emptyWaitSet.signal();
  193.         }finally {
  194.             lock.unlock();
  195.         }
  196.     }
  198.     //带超时时间的阻塞添加
  199.     public boolean offer(T task, long timeout, TimeUnit timeUnit){
  200.         lock.lock();
  201.         try {
  202.             long nanos = timeUnit.toNanos(timeout);
  203.             while (queue.size()==capcity){
  204.                 try {
  205.                     nanos = fuulWaitSet.awaitNanos(nanos);
  206.                 } catch (InterruptedException e) {
  207.                     e.printStackTrace();
  208.                 }
  209.             }
  210.             queue.addLast(task);
  211.             emptyWaitSet.signal();
  212.         }finally {
  213.             lock.unlock();
  214.         }
  215.     }
  217.     //获取大小,当然这里可以不加锁，完全只读
  218.     public int size(){
  219.         lock.lock();
  220.         try {
  221.             return queue.size();
  222.         }finally {
  223.             lock.unlock();
  224.         }
  225.     }
  227.     public void tryPut(RejectPolicy<T> rejectPolicy, T task){
  228.         lock.lock();
  229.         try {
  230.             //判断队列是否为满
  231.             if(queue.size()==capcity){
  232.                 //传入就好，权力下放给调用者，妙！
  233.                 rejectPolicy.reject(this,task);
  234.             }else{
  235.                 queue.addLast(task);
  236.             }
  237.         }
  238.         finally {
  240.         }
  241.     }
  243. }

（附）补充知识-函数式接口与Lambda表达式：

Java8 函数式接口，函数式接口（Functional Interface）是一个有且仅有一个抽象方法，但是可以有多个非抽象方法的接口。
函数式接口可以被隐式转换为lambda表达式。

如下定义了一个函数式接口：

  1. @FunctionalInterface
  2. interface GreetingService{
  3.     void sayMessage(String message);
  4. }

接口中只有一个抽象方法，该抽象方法的参数是 String message。

使用Lambda表达式来表示该接口的一个实现类对象：

  1. GreetingService greetService1 = message->System.out.println("hello "+ message);

其中message是待实现方法的参数，当有多个参数时用小括号扩起，->后是方法的实现细节

（附）补充知识-接口实现类对象的快捷创建：

与函数式接口+lambda表达式的用法类似，当需要快捷创建一个接口的实现类对象时，无需先定义接口实现类，接着再new impl()对象，而可以直接使用如下语法：

  1. new interface (){
  2.     // 接口实现类细节...
  4.     @Override 
  5.     // 实现方法...
  6. }

示例代码：

  1. public class TestInterfaceImpl {
  2.     public static void main(String[] args) {
  3.         Animal animal = new Animal() {
  4.             @Override
  5.             public void run() {
  6.                 System.out.println("I am a dog and I can fly");
  7.             }
  8.         };
  9.     }
  10. }
  12. interface Animal{
  13.     void run();
  14. }