package cscore.concurrent.java.threadpool;
 
import java.util.concurrent.BlockingQueue;
 
public class Worker implements Runnable {
 
  private Thread thisThread; // 表示正在執(zhí)行任務的線程
  private BlockingQueue<Runnable> taskQueue; // 由線程池傳遞過來的任務隊列
  private volatile boolean isStopped; // 表示 worker 是否停止工作 需要使用 volatile 保證線程之間的可見性
 
  public Worker(BlockingQueue taskQueue) { // 這個構造方法是在線程池的實現(xiàn)當中會被調用
    this.taskQueue = taskQueue;
  }
 
  // 線程執(zhí)行的函數(shù)
  @Override
  public void run() {
    thisThread = Thread.currentThread(); // 獲取執(zhí)行任務的線程
    while (!isStopped) { // 當線程沒有停止的時候就不斷的去任務池當中取出任務
      try {
        Runnable task = taskQueue.take(); // 從任務池當中取出任務 當沒有任務的時候線程會被這個方法阻塞
        task.run(); // 執(zhí)行任務 任務就是一個 Runnable 對象
      } catch (InterruptedException e) {
        // do nothing
        // 這個地方很重要 你有沒有思考過一個問題當任務池當中沒有任務的時候 線程會被阻塞在 take 方法上
        // 如果我們后面沒有任務提交拿他就會一直阻塞 那么我們該如何喚醒他呢
        // 答案就在下面的函數(shù)當中 調用線程的 interruput 方法 那么take方法就會產(chǎn)生一個異常 然后我們
        // 捕獲到一異常 然后線程退出
      }
    }
  }
 
  public synchronized void stopWorker() {
    if (isStopped) {
      throw new RuntimeException("thread has been interrupted");
    }
    isStopped = true;
    thisThread.interrupt(); // 中斷線程產(chǎn)生異常
  }
 
  public synchronized boolean isStopped() {
    return isStopped;
  }
}

  private BlockingQueue taskQueue; // 任務池
  private volatile boolean isStopped; // 
  private final List<Worker> workers = new ArrayList<>();// 保存所所有的執(zhí)行任務的線程
 
  public ThreadPool(int numThreads, int maxTasks) {
    this.taskQueue = new ArrayBlockingQueue(maxTasks);
    for (int i = 0; i < numThreads; i++) {
      workers.add(new Worker(this.taskQueue));
    }
    int i = 1;
    // 這里產(chǎn)生線程 然后啟動線程
    for (Worker worker : workers) {
      new Thread(worker, "ThreadPool-" + i + "-thread").start();
      i++;
    }
  }

package cscore.concurrent.java.threadpool;
 
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
 
public class ThreadPool {
 
  private BlockingQueue taskQueue;
  private volatile boolean isStopped;
  private final List<Worker> workers = new ArrayList<>();
 
  public ThreadPool(int numThreads, int maxTasks) {
    this.taskQueue = new ArrayBlockingQueue(maxTasks);
    for (int i = 0; i < numThreads; i++) {
      workers.add(new Worker(this.taskQueue));
    }
    int i = 1;
    for (Worker worker : workers) {
      new Thread(worker, "ThreadPool-" + i + "-thread").start();
      i++;
    }
  }
 
  // 下面這個方法是向線程池提交任務
  public void execute(Runnable runnable) throws InterruptedException {
    if (isStopped) {
      // 如果線程池已經(jīng)停下來了，就不在向任務隊列當中提交任務了
      System.err.println("thread pool has been stopped, so quit submitting task");
      return;
    }
    taskQueue.put(runnable);
  }
 
  // 強制關閉線程池
  public synchronized void stop() {
    isStopped = true;
    for (Worker worker : workers) {
      worker.stopWorker();
    }
  }
 
  public synchronized void shutDown() {
    // 先表示關閉線程池 線程就不能再向線程池提交任務
    isStopped = true;
    // 先等待所有的任務執(zhí)行完成再關閉線程池
    waitForAllTasks();
    stop();
  }
 
  private void waitForAllTasks() {
    // 當線程池當中還有任務的時候 就不退出循環(huán)
    while (taskQueue.size() > 0)
      Thread.yield();
  }
}

package cscore.concurrent.java.threadpool;
 
public class TestPool {
 
  public static void main(String[] args) throws InterruptedException {
    ThreadPool pool = new ThreadPool(3, 1024);
 
    for (int i = 0; i < 10; i++) {
      int tmp = i;
      pool.execute(() -> {
        System.out.println(Thread.currentThread().getName() + " say hello " + tmp);
      });
    }
    pool.shutDown();
  }
}

    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == 0)
                notEmpty.await();
            return dequeue();
        } finally {
            lock.unlock();
        }
    }