Java基础—线程池原理与使用

public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
            BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
if (corePoolSize < 0 ||maximumPoolSize <= 0 ||maximumPoolSize < corePoolSize ||keepAliveTime < 0)
            throw new IllegalArgumentException();
if (workQueue == null || threadFactory == null || handler == null)
      throw new NullPointerException();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

 public static void main(String[] args) {
        Runnable task = new Runnable() {
                @Override
                public void run()
 {
                    System.out.println("run a task");
                }
        };
      //定义一个有10个线程的线程池
         ExecutorService fixedThreadPool = Executors.newFixedThreadPool(10); 
      fixedThreadPool.submit(task)
         //定义一个只有一个线程的线程池
        ExecutorService singleThreadPool = Executors.newSingleThreadExecutor();
       singleThreadPool.submit(task)
        //定义一个线程数可伸缩的线程池
        ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
        cachedThreadPool.submit(task)
        //定义一个可拆分任务的线程池
        ExecutorService workStealPool = Executors.newWorkStealingPool();
  }

//ExecutorService核心方法
public Future<?> submit(Runnable task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<Void> ftask = newTaskFor(task, null);
        execute(ftask);
        return ftask;
}

//ThreadPoolExecutor核心方法
public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        int c = ctl.get(); //获取线程池的状态
        //1、判断当前空闲线程数是否小于核心线程数，若是则新建核心线程并把任务设为首个任务
        if (workerCountOf(c) < corePoolSize) {
            //若addWorker返回false直接结束，addWork方法将会原子性地检查runState、workCount
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        //2、判断线程池是否处于运行状态及是否成功添加任务到队列中
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            /*
            2.1 重新检查状态，有可能最后一次检查时存在的线程已被销毁或进入此方法时线程池被关闭
            成功添加任务后双重检查线程池的状态，若线程池状态为STOP，则移除队列中的任务并执行拒绝策略
            */
            if (! isRunning(recheck) && remove(command))
                reject(command);
            //2.2 线程池处于运行状态且当前空闲线程数为零则创建非核心线程
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        //3、如果添加任务失败则尝试创建非核心线程，若创建失败，有可能线程池已被关闭或队列已饱和，拒绝任务
        else if (!addWorker(command, false))
            reject(command);
}

//添加线程
private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        //CAS方式添加线程
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
      //线程池处于非运行状态 and !(线程池处于关闭状态 and 任务为空 and 任务队列非空)
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;
       //通过添加
            for (;;) {
                int wc = workerCountOf(c);
                //判断当前线程数是否大于最大线程数限制，若超过则返回false
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                //通过cas更新当前线程数+1
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                c = ctl.get();  // Re-read ctl
                //若cas操作失败则重新读取当前线程数,并判断当前线程池的状态是否被更新
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
        }

        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;
                //获取锁并添加新的线程到workers集合中
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    int rs = runStateOf(ctl.get());
          //由于ThreadFactory创建线程时失败或获到锁前线程池已被SHUTDOWN，故重新检查线程池状态
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable(判断创建的线程是否存活)
                            throw new IllegalThreadStateException();
                        workers.add(w);
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        //添加成功后更新添加标志位
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                //线程创建成功后，启动线程并更新启动标志位
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            //若线程启动或创建失败，执行失败逻辑
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
}

private void addWorkerFailed(Worker w) {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (w != null)
                workers.remove(w);  //从集合中移除线程
            decrementWorkerCount(); //原子性更新线程数
            tryTerminate();  // 尝试终止线程池
        } finally {
            mainLock.unlock();
        }
}