讲解+源码分析+实战，近万字讲解线程池，不懂都不行！

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)

public class SimpleApplication {


    public static void main(String[] args) {


        ThreadPoolExecutor pool = new ThreadPoolExecutor(4, 10, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100));
        Future<List<Integer>> future1 = pool.submit(new Counter(100));
        Future<List<Integer>> future2 = pool.submit(new Counter(50));


        try {
            List<Integer> result1 = future1.get();
            List<Integer> result2 = future2.get();


            System.out.println("得到的结果1:" + "大小:" + result1.size() + ",生成的数据:" + result1);
            System.out.println("得到的结果2:" + "大小:" + result2.size() + ",生成的数据:" + result2);
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }


        pool.shutdown();
    }


    /**
     * 生成特定个数的随机数
     */
    static class Counter implements Callable<List<Integer>> {


        private final int size;


        public Counter(int size) {
            this.size = size;
        }


        @Override
        public List<Integer> call() throws Exception {
            if (size < 0) {
                return new ArrayList<>();
            }


            ArrayList<Integer> arrayList = new ArrayList<>(size);


            Random random = new Random();
            for (int i = 0; i < size ; i++) {
                arrayList.add(random.nextInt());
            }


            return arrayList;
        }
    }
}

List<Callable<T>> tasks = ...;
List<Future<T>> results = pool.invokeAll(tasks);
for (Future<T> result : results) {
  result.get();
}

ExecutorCompletionService<T> service = new ExecutorCompletionService<>(pool);
for (Callable<T> task : tasks) {
  service.submit(task);
}


for (int i = 0; i < tasks.size(); i++) {
  service.take().get();
}

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

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.
         *
         * 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.
         */
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        else if (!addWorker(command, false))
            reject(command);
    }

public boolean offer(E e) {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            if (count == items.length)
                return false;
            else {
                enqueue(e);
                return true;
            }
        } finally {
            lock.unlock();
        }
    }

private boolean addWorker(Runnable firstTask, boolean core) {
        //对代码块进行标记
        retry:
        for (;;) {
            int c = ctl.get();
            //获取高3位的线程运行状态
            int rs = runStateOf(c);


            // Check if queue empty only if necessary.
            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;
                //线程计数执行原子加1操作，成功直接跳出最外面那层循环
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                c = ctl.get();  // Re-read ctl
                //状态改变了，跳出内层循环，重新执行外层循环，重新获取状态
                if (runStateOf(c) != rs)
                    continue retry;
                //状态未改变，一直在内层循环重试CAS加1操作
                // else CAS failed due to workerCount change; retry inner loop
            }
        }


        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            //将任务封装成Worker对象
            w = new Worker(firstTask);
            //获取到Worker对象创建的Thread对象
            final Thread t = w.thread;
            if (t != null) {
                //上锁，保证线程安全
                final ReentrantLock mainLock = this.mainLock;                
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    int rs = runStateOf(ctl.get());
                    //线程池处于运行状态   或   线程池处于关闭状态且task为null
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        //添加到任务队列，任务队列用Set存储
                        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;
    }

final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
            //while循环，当firstTask执行完成后或本来就为空时
            //会通过getTask方法从阻塞队列中获取新的任务执行
            //如果getTask()结果为null，则跳出循环，结束线程运行
            while (task != null || (task = getTask()) != null) {
                w.lock();
                // If pool is stopping, ensure thread is interrupted;
                // if not, ensure thread is not interrupted.  This
                // requires a recheck in second case to deal with
                // shutdownNow race while clearing interrupt
                if ((runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                    wt.interrupt();
                try {
                    //默认什么都不做
                    //可以通过重写此方法实现扩展
                    beforeExecute(wt, task);
                    Throwable thrown = null;
                    try {
                        //真正执行的代码块
                        //这里是在submit方法上封装的FutureTask对象的run方法
                        task.run();
                    } catch (RuntimeException x) {
                        thrown = x; throw x;
                    } catch (Error x) {
                        thrown = x; throw x;
                    } catch (Throwable x) {
                        thrown = x; throw new Error(x);
                    } finally {
                        //默认什么都不做
                        //可以通过重写此方法实现扩展
                        afterExecute(task, thrown);
                    }
                } finally {
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }

private Runnable getTask() {
        boolean timedOut = false; // Did the last poll() time out?


        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);


            // Check if queue empty only if necessary.
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }


            int wc = workerCountOf(c);


            // Are workers subject to culling?
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;


            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }


            try {
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }

public class Application {


    private static final ExecutorService pool = new ThreadPoolExecutor(10, 20, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100));


    public static void main(String[] args) {


        try (Scanner in = new Scanner(System.in)) {
            System.out.println("请输入要检索的目录:");
            String directory = in.nextLine();
            System.out.println("请输入要检索的关键字:");
            String keyword = in.nextLine();


            MatchCounter counter = new MatchCounter(new File(directory), keyword);


            Future<Integer> future = pool.submit(counter);




            System.out.println(future.get() + "个匹配的文件");
            System.out.println("largest pool size:" + ((ThreadPoolExecutor)pool).getLargestPoolSize());


            pool.shutdown();
        } catch (Exception e) {
            e.printStackTrace();
        }




    }


    /**
     * 搜索关键字出现的文件数
     */
    static class MatchCounter implements Callable<Integer> {


        private final File directory;
        private final String keyword;


        public MatchCounter(File directory, String keyword) {
            this.directory = directory;
            this.keyword = keyword;
        }


        @Override
        public Integer call() {


            int count = 0;


            File[] files = directory.listFiles();


            if (files == null) {
                return 0;
            }
            List<Future<Integer>> results = new ArrayList<>();


            for (File file : files) {
                if (file.isDirectory()) {
                    MatchCounter counter = new MatchCounter(file, keyword);


                    Future<Integer> future = pool.submit(counter);
                    results.add(future);
                } else {
                    if (search(file)) {
                        count++;
                    }
                }
            }


            for (Future<Integer> result : results) {
                try {
                    count += result.get();
                } catch (InterruptedException | ExecutionException e) {
                    e.printStackTrace();
                }
            }


            return count;
        }


        private boolean search(File file) {


            try (Scanner in = new Scanner(file, "UTF-8")) {
                boolean found = false;
                while (in.hasNextLine()) {
                    String line = in.nextLine();
                    if (line.contains(keyword)) {
                        System.out.println("检索到的文件:" + file.getName());
                        return true;
                    }
                }
                return false;
            } catch (IOException e) {
                return false;
            }
        }
    }


}

public class ScheduledApplication {


    private static final ScheduledThreadPoolExecutor sPool = new ScheduledThreadPoolExecutor(4);
    public static void main(String[] args) {


        Scanner scanner = new Scanner(System.in);
        //每1秒执行一次
        sPool.scheduleAtFixedRate(new TimeReport(), 1000, 1000, TimeUnit.MILLISECONDS);
        System.out.println("2s后执行预期任务:" + new Date());
        sPool.schedule(new FutureTask(), 2000, TimeUnit.MILLISECONDS);


        if (scanner.hasNext()) {
            sPool.shutdown();
        }
    }


    static class TimeReport implements Runnable {


        @Override
        public void run() {
            System.out.println(new Date());
        }
    }


    static class FutureTask implements Runnable {


        @Override
        public void run() {
            System.out.println("预期执行:" + new Date());
        }
    }
}