Java实现手写乞丐版线程池的示例代码

package cscore.concurrent.java.threadpool;
 
import java.util.concurrent.BlockingQueue;
 
public class Worker implements Runnable {
 
  private Thread thisThread; // 表示正在执行任务的线程
  private BlockingQueue<Runnable> taskQueue; // 由线程池传递过来的任务队列
  private volatile boolean isStopped; // 表示 worker 是否停止工作 需要使用 volatile 保证线程之间的可见性
 
  public Worker(BlockingQueue taskQueue) { // 这个构造方法是在线程池的实现当中会被调用
    this.taskQueue = taskQueue;
  }
 
  // 线程执行的函数
  @Override
  public void run() {
    thisThread = Thread.currentThread(); // 获取执行任务的线程
    while (!isStopped) { // 当线程没有停止的时候就不断的去任务池当中取出任务
      try {
        Runnable task = taskQueue.take(); // 从任务池当中取出任务 当没有任务的时候线程会被这个方法阻塞
        task.run(); // 执行任务 任务就是一个 Runnable 对象
      } catch (InterruptedException e) {
        // do nothing
        // 这个地方很重要 你有没有思考过一个问题当任务池当中没有任务的时候 线程会被阻塞在 take 方法上
        // 如果我们后面没有任务提交拿他就会一直阻塞 那么我们该如何唤醒他呢
        // 答案就在下面的函数当中 调用线程的 interruput 方法 那么take方法就会产生一个异常 然后我们
        // 捕获到一异常 然后线程退出
      }
    }
  }
 
  public synchronized void stopWorker() {
    if (isStopped) {
      throw new RuntimeException("thread has been interrupted");
    }
    isStopped = true;
    thisThread.interrupt(); // 中断线程产生异常
  }
 
  public synchronized boolean isStopped() {
    return isStopped;
  }
}

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

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) {
      // 如果线程池已经停下来了，就不在向任务队列当中提交任务了
      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;
    // 先等待所有的任务执行完成再关闭线程池
    waitForAllTasks();
    stop();
  }
 
  private void waitForAllTasks() {
    // 当线程池当中还有任务的时候 就不退出循环
    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();
        }
    }