package cscore.concurrent.java.threadpool;
 
import java.util.concurrent.BlockingQueue;
 
public class Worker implements Runnable {
 
  private Thread thisThread; // 表示正在執(zhí)行任務(wù)的線程
  private BlockingQueue<Runnable> taskQueue; // 由線程池傳遞過(guò)來(lái)的任務(wù)隊(duì)列
  private volatile boolean isStopped; // 表示 worker 是否停止工作 需要使用 volatile 保證線程之間的可見(jiàn)性
 
  public Worker(BlockingQueue taskQueue) { // 這個(gè)構(gòu)造方法是在線程池的實(shí)現(xiàn)當(dāng)中會(huì)被調(diào)用
    this.taskQueue = taskQueue;
  }
 
  // 線程執(zhí)行的函數(shù)
  @Override
  public void run() {
    thisThread = Thread.currentThread(); // 獲取執(zhí)行任務(wù)的線程
    while (!isStopped) { // 當(dāng)線程沒(méi)有停止的時(shí)候就不斷的去任務(wù)池當(dāng)中取出任務(wù)
      try {
        Runnable task = taskQueue.take(); // 從任務(wù)池當(dāng)中取出任務(wù) 當(dāng)沒(méi)有任務(wù)的時(shí)候線程會(huì)被這個(gè)方法阻塞
        task.run(); // 執(zhí)行任務(wù) 任務(wù)就是一個(gè) Runnable 對(duì)象
      } catch (InterruptedException e) {
        // do nothing
        // 這個(gè)地方很重要 你有沒(méi)有思考過(guò)一個(gè)問(wèn)題當(dāng)任務(wù)池當(dāng)中沒(méi)有任務(wù)的時(shí)候 線程會(huì)被阻塞在 take 方法上
        // 如果我們后面沒(méi)有任務(wù)提交拿他就會(huì)一直阻塞 那么我們?cè)撊绾螁拘阉?
        // 答案就在下面的函數(shù)當(dāng)中 調(diào)用線程的 interruput 方法 那么take方法就會(huì)產(chǎn)生一個(gè)異常 然后我們
        // 捕獲到一異常 然后線程退出
      }
    }
  }
 
  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; // 任務(wù)池
  private volatile boolean isStopped; // 
  private final List<Worker> workers = new ArrayList<>();// 保存所所有的執(zhí)行任務(wù)的線程
 
  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)生線程 然后啟動(dòng)線程
    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++;
    }
  }
 
  // 下面這個(gè)方法是向線程池提交任務(wù)
  public void execute(Runnable runnable) throws InterruptedException {
    if (isStopped) {
      // 如果線程池已經(jīng)停下來(lái)了，就不在向任務(wù)隊(duì)列當(dāng)中提交任務(wù)了
      System.err.println("thread pool has been stopped, so quit submitting task");
      return;
    }
    taskQueue.put(runnable);
  }
 
  // 強(qiáng)制關(guān)閉線程池
  public synchronized void stop() {
    isStopped = true;
    for (Worker worker : workers) {
      worker.stopWorker();
    }
  }
 
  public synchronized void shutDown() {
    // 先表示關(guān)閉線程池 線程就不能再向線程池提交任務(wù)
    isStopped = true;
    // 先等待所有的任務(wù)執(zhí)行完成再關(guān)閉線程池
    waitForAllTasks();
    stop();
  }
 
  private void waitForAllTasks() {
    // 當(dāng)線程池當(dāng)中還有任務(wù)的時(shí)候 就不退出循環(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();
        }
    }