并发编程技术是互联网应用开发中必须掌握的知识，本篇文章笔者开始分析线程池，进一步了解Java领域并发编程知识。

上文Java并发编程之Thread知识整理已经介绍了线程的目的和好处，线程池就是为了更好的使用线程，发挥线程最大的价值。

线程池创建的方式有3种：

第一种通过ThreadPoolExecutor的构造函数方法创建。通过ThreadPoolExecutor的最全构造参数来看总共有7个参数。

public ThreadPoolExecutor(    int corePoolSize, //核心线程数大小    int maximumPoolSize, //最大线程数    long keepAliveTime, // 超过核心线程数后的线程存活时间    TimeUnit unit, //存活时间单位    BlockingQueue<Runnable> workQueue, //任务队列    ThreadFactory threadFactory, // 创建线程的工厂     RejectedExecutionHandler handler // 拒绝策略处理器            ) {}

第二种，ScheduledThreadPoolExecutor类的构造器，创建定时，或者一定频率执行的线程池。

 public ScheduledThreadPoolExecutor(     int corePoolSize,//核心线程数     ThreadFactory threadFactory,//创建线程的工厂     RejectedExecutionHandler handler //池满拒绝策略              )        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,              new DelayedWorkQueue(), threadFactory, handler);    }
public ScheduledFuture<?> scheduleAtFixedRate(                Runnable command, //定时执行的任务                long initialDelay,//首次延迟启动时间                long period,//定期周期                TimeUnit unit//定期周期单位                                    ) {        if (command == null || unit == null)            throw new NullPointerException();        if (period <= 0)            throw new IllegalArgumentException();        ScheduledFutureTask<Void> sft =            new ScheduledFutureTask<Void>(command,null,                            triggerTime(initialDelay, unit),  unit.toNanos(period));        RunnableScheduledFuture<Void> t = decorateTask(command, sft);        sft.outerTask = t;        delayedExecute(t);        return t;    }

第三种使用工厂Executors类的静态方法

//弹性可伸缩的线程池，核心线程大小为0，最大线程数为Integer.MAX_VALUE,只要小于Integer.MAX_VALUE就会创建线程加入线程池，只要线程空余时间大于60s则回收线程资源。可能导致线程创建过多，耗尽系统资源，使用同步队列，不存储任务，没有任务堆积能力，强依赖线程资源public static ExecutorService newCachedThreadPool() {    return new ThreadPoolExecutor(    0,     Integer.MAX_VALUE,    60L,     TimeUnit.SECONDS,    new SynchronousQueue<Runnable>());}

// 固定线程大小的线程池，核心线程数和最大线程数都是一样的，//队列是无界的，只要线程数小于核心线程数，就加线程，//只要大于核心线程数，就加入无界队列。线程大小固定，//如果任务执行时间长，可能导致任务堆积过久。 public static ExecutorService newFixedThreadPool(int nThreads) {        return new ThreadPoolExecutor(nThreads,         nThreads,         0L,         TimeUnit.MILLISECONDS,         new LinkedBlockingQueue<Runnable>());  }

// 线程大小只有1的线程池，队列采用有界阻塞队列，任务会按提交顺序执行public static ExecutorService newSingleThreadExecutor() {        return new FinalizableDelegatedExecutorService            (new ThreadPoolExecutor(1, 1,                                    0L, TimeUnit.MILLISECONDS,                                    new LinkedBlockingQueue<Runnable>()));    }

//创建单个定时调度的线程池，这个在中间件中用的比较多，比如rocketmq,dubbo//用于后台执行一些定时操作。public static ScheduledExecutorService newSingleThreadScheduledExecutor() {return new DelegatedScheduledExecutorService            (new ScheduledThreadPoolExecutor(1));    }

在阿里巴巴出品的Java开发手册中，明确强制不允许使用Executors类创建线程池，因为通过Executors存在耗尽资源的风险。

线程池原理

线程池的存储结构：线程封装在Worker类中，并使用HashSet存储池中的线程

     /**     * Set containing all worker threads in pool. Accessed only when     * holding mainLock.     */  private final HashSet<Worker> workers = new HashSet<Worker>();

提交任务到线程池：

java.util.concurrent.ThreadPoolExecutor#execute(Runnable r)流程



 // 1、如果当前正在运行的线程数小于核心线程数，启动一个核心线程，将r作为该线程第一个执行的任务//2、如果核心线程数满了，将任务加入任务队列//3、如果任务队列满了，增加非核心线程数，直到达到最大线程数//4、如果达到了最大线程数，则执行拒绝策略RejectedExecutionHandler // AbortPolicy（抛异常，默认）DiscardPolicy(空处理，丢弃) // DiscardOldestPolicy(丢弃(poll)任务队列中的下一个将执行任务-头部任务，并尝试执行当前任务) // CallerRunsPolicy (调用者线程直接调用run方法)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);    }

线程池也可以实现预热功能，将核心线程数，预先创建好

java.util.concurrent.ThreadPoolExecutor#prestartCoreThread java.util.concurrent.ThreadPoolExecutor#ensurePrestart  //至少启动一个线程 java.util.concurrent.ThreadPoolExecutor#prestartAllCoreThreads   // 启动所有核心线程                       
//控制核心线程的空闲时间 java.util.concurrent.ThreadPoolExecutor#allowCoreThreadTimeOut //开启核心线程回收功能（默认不回收）//提交任务2 java.util.concurrent.AbstractExecutorService#submit(java.util.concurrent.Callable<T>)     // 也是调用了execute 只不过将任务构造成了FutureTask                java.util.concurrent.FutureTask#FutureTask(java.util.concurrent.Callable<V>)FutureTask实现了Runnable接口的run方法，并且持有了Callable的引用. 提交到线程池的是FutureTask任务。// 线程池会调用这个run方法public void run() {        if (state != NEW ||            !RUNNER.compareAndSet(this, null, Thread.currentThread()))            return;        try {            Callable<V> c = callable;            if (c != null && state == NEW) {                V result;                boolean ran;                try {                //调用call方法                    result = c.call();                    ran = true;                } catch (Throwable ex) {                    result = null;                    ran = false;                    setException(ex);                }                if (ran)                //设置完成状态和结果 Locksupport.unpark对等待结果的线程进行唤醒                    set(result);            }        } finally {            // runner must be non-null until state is settled to            // prevent concurrent calls to run()            runner = null;            // state must be re-read after nulling runner to prevent            // leaked interrupts            int s = state;            if (s >= INTERRUPTING)                handlePossibleCancellationInterrupt(s);        }    }

线程池提交任务逻辑流程图

执行任务原理：

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 (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 {                    //调用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);        }    }

任务包装类Worker ：

//继承了aqs 每个任务执行前都需要加锁private final class Worker        extends AbstractQueuedSynchronizer        implements Runnable{        /**         * This class will never be serialized, but we provide a         * serialVersionUID to suppress a javac warning.         */        private static final long serialVersionUID = 6138294804551838833L;
        /** Thread this worker is running in.  Null if factory fails. */        final Thread thread;        /** Initial task to run.  Possibly null. */        Runnable firstTask;        /** Per-thread task counter */        volatile long completedTasks;
        /**         * Creates with given first task and thread from ThreadFactory.         * @param firstTask the first task (null if none)         */        Worker(Runnable firstTask) {            setState(-1); // inhibit interrupts until runWorker            this.firstTask = firstTask;            this.thread = getThreadFactory().newThread(this);        }
        /** Delegates main run loop to outer runWorker  */        public void run() {            runWorker(this);        }
        // Lock methods        //        // The value 0 represents the unlocked state.        // The value 1 represents the locked state.
        protected boolean isHeldExclusively() {            return getState() != 0;        }
        protected boolean tryAcquire(int unused) {            if (compareAndSetState(0, 1)) {                setExclusiveOwnerThread(Thread.currentThread());                return true;            }            return false;        }
        protected boolean tryRelease(int unused) {            setExclusiveOwnerThread(null);            setState(0);            return true;        }
        public void lock()        { acquire(1); }        public boolean tryLock()  { return tryAcquire(1); }        public void unlock()      { release(1); }        public boolean isLocked() { return isHeldExclusively(); }
        void interruptIfStarted() {            Thread t;            if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {                try {                    t.interrupt();                } catch (SecurityException ignore) {                }            }        }    }