由浅入深理解Java线程池及线程池的如何使用详解编程语言

 public ThreadPoolExecutor(int paramInt1, int paramInt2, long paramLong, TimeUnit paramTimeUnit, BlockingQueue<Runnable> paramBlockingQueue, ThreadFactory paramThreadFactory, RejectedExecutionHandler paramRejectedExecutionHandler) { this.ctl = new AtomicInteger(ctlOf(-536870912, 0)); this.mainLock = new ReentrantLock(); this.workers = new HashSet(); this.termination = this.mainLock.newCondition(); if ((paramInt1 < 0) || (paramInt2 <= 0) || (paramInt2 < paramInt1) || (paramLong < 0L)) throw new IllegalArgumentException(); if ((paramBlockingQueue == null) || (paramThreadFactory == null) || (paramRejectedExecutionHandler == null)) throw new NullPointerException(); this.corePoolSize = paramInt1; this.maximumPoolSize = paramInt2; this.workQueue = paramBlockingQueue; this.keepAliveTime = paramTimeUnit.toNanos(paramLong); this.threadFactory = paramThreadFactory; this.handler = paramRejectedExecutionHandler; }

    public void execute(Runnable command) { 
        if (command == null) 
            throw new NullPointerException(); 
        /* 
         * Proceed in 3 steps: 
         * 
         * 1. If fewer than corePoolSize threads are running, try to 
         * start a new thread with the given command as its first 
         * task.  The call to addWorker atomically checks runState and 
         * workerCount, and so prevents false alarms that would add 
         * threads when it shouldn't, by returning false. 
         * 如果正在运行的线程数小于corePoolSize，那么将调用addWorker 方法来创建一个新的线程，并将该任务作为新线程的第一个任务来执行。
　　　　　　 当然，在创建线程之前会做原子性质的检查，如果条件不允许，则不创建线程来执行任务，并返回false.　　
 
         * 2. If a task can be successfully queued, then we still need 
         * to double-check whether we should have added a thread 
         * (because existing ones died since last checking) or that 
         * the pool shut down since entry into this method. So we 
         * recheck state and if necessary roll back the enqueuing if 
         * stopped, or start a new thread if there are none. 
         * 如果一个任务成功进入阻塞队列，那么我们需要进行一个双重检查来确保是我们已经添加一个线程（因为存在着一些线程在上次检查后他已经死亡）或者
　　　　　　 当我们进入该方法时，该线程池已经关闭。所以，我们将重新检查状态，线程池关闭的情况下则回滚入队列，线程池没有线程的情况则创建一个新的线程。 
         * 3. If we cannot queue task, then we try to add a new 
         * thread.  If it fails, we know we are shut down or saturated 
         * and so reject the task.
　　　　　　 如果任务无法入队列（队列满了），那么我们将尝试新开启一个线程（从corepoolsize到扩充到maximum），如果失败了，那么可以确定原因，要么是
　　　　　　 线程池关闭了或者饱和了（达到maximum），所以我们执行拒绝策略。
 
         */
　　　　
　　　　// 1.当前线程数量小于corePoolSize，则创建并启动线程。 
        int c = ctl.get(); 
        if (workerCountOf(c) < corePoolSize) { 
            if (addWorker(command, true))
　　　　　　　　// 成功，则返回

return; 
            c = ctl.get(); 
        }
　　　　// 2.步骤1失败，则尝试进入阻塞队列， 
        if (isRunning(c) && workQueue.offer(command)) {
　　　　　　　// 入队列成功，检查线程池状态，如果状态部署RUNNING而且remove成功，则拒绝任务 
            int recheck = ctl.get(); 
            if (! isRunning(recheck) && remove(command)) 
                reject(command);
　　　　　　　// 如果当前worker数量为0，通过addWorker(null, false)创建一个线程，其任务为null 
            else if (workerCountOf(recheck) == 0) 
                addWorker(null, false); 
        }
　　　　// 3. 步骤1和2失败，则尝试将线程池的数量有corePoolSize扩充至maxPoolSize，如果失败，则拒绝任务 
        else if (!addWorker(command, false)) 
            reject(command); 
    }

private boolean addWorker(Runnable firstTask, boolean core) { 
        retry:
　　　　 // 外层循环，用于判断线程池状态 
        for (;;) { 
            int c = ctl.get(); 
            int rs = runStateOf(c); 
 
            // Check if queue empty only if necessary. 
            if (rs >= SHUTDOWN && 
                ! (rs == SHUTDOWN && 
                   firstTask == null && 
                   ! workQueue.isEmpty())) 
                return false; 
　　　　　　 // 内层的循环，任务是将worker数量加1 
            for (;;) { 
                int wc = workerCountOf(c); 
                if (wc >= CAPACITY || 
                    wc >= (core ? corePoolSize : maximumPoolSize)) 
                    return false; 
                if (compareAndIncrementWorkerCount(c)) 
                    break retry; 
                c = ctl.get();  // Re-read ctl 
                if (runStateOf(c) != rs) 
                    continue retry; 
                // else CAS failed due to workerCount change; retry inner loop 
            } 
        } 
　　　　// worker加1后，接下来将woker添加到HashSet<Worker>中，并启动worker 
        boolean workerStarted = false; 
        boolean workerAdded = false; 
        Worker w = null; 
        try { 
            final ReentrantLock mainLock = this.mainLock; 
            w = new Worker(firstTask); 
            final Thread t = w.thread; 
            if (t != null) { 
                mainLock.lock(); 
                try { 
                    // Recheck while holding lock. 
                    // Back out on ThreadFactory failure or if 
                    // shut down before lock acquired. 
                    int c = ctl.get(); 
                    int rs = runStateOf(c); 
 
                    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(); 
                }
　　　　　　　　　// 如果往HashSet<Worker>添加成功，则启动该线程 
                if (workerAdded) { 
                    t.start(); 
                    workerStarted = true; 
                } 
            } 
        } finally { 
            if (! workerStarted) 
                addWorkerFailed(w); 
        } 
        return workerStarted; 
    }

public static ExecutorService newFixedThreadPool(int var0) { 
        return new ThreadPoolExecutor(var0, var0, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue()); 
    }

public static ExecutorService newFixedThreadPool(int var0, ThreadFactory var1) {
    return new ThreadPoolExecutor(var0, var0, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue(), var1);
}

public static ExecutorService newSingleThreadExecutor() { 
        return new Executors.FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue())); 
    } 
 
    public static ExecutorService newSingleThreadExecutor(ThreadFactory var0) { 
        return new Executors.FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue(), var0)); 
    }

public static ExecutorService newCachedThreadPool() { 
        return new ThreadPoolExecutor(0, 2147483647, 60L, TimeUnit.SECONDS, new SynchronousQueue()); 
    } 
 
    public static ExecutorService newCachedThreadPool(ThreadFactory var0) { 
        return new ThreadPoolExecutor(0, 2147483647, 60L, TimeUnit.SECONDS, new SynchronousQueue(), var0); 
    }

public static ScheduledExecutorService newScheduledThreadPool(int var0) { 
        return new ScheduledThreadPoolExecutor(var0); 
    } 
 
    public static ScheduledExecutorService newScheduledThreadPool(int var0, ThreadFactory var1) { 
        return new ScheduledThreadPoolExecutor(var0, var1); 
    }

int ncpus = Runtime.getRuntime().availableProcessors();

    private static ExecutorService createDefaultExecutorService(Args args) { 
        SynchronousQueue executorQueue = new SynchronousQueue(); 
 
        return new ThreadPoolExecutor(args.minWorkerThreads, args.maxWorkerThreads, 60L, TimeUnit.SECONDS, 
                executorQueue); 
    }