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);});}}
}

?因為調用了BlockingQueue的take方法，當隊列為空的時候，會無限循環等待，所以這兩個線程一直沒有結束。可以調用帶超時的poll方法，超時后，線程就會結束，也從線程集合中移除。