ThreadPoolExecutor是一个ExecutorService,实现了Executor接口,是一个线程池服务的实现。通常通过Executors类的工厂方法来创建。

一、实现原理

1、关键要素

要理解线程池的实现原理，主要理解以下的关键要素：

• 线程池的状态：提供了线程池生命周期中所处的不同状态，分别为以下状态(分别用0，1，2，3表示)：
①RUNNING: 可以接收新任务和可以处理队列中的任务；
②SHUTDOWN: 不接收新任务，可以继续处理队列中的任务；
③STOP: 不接收新任务，也不处理队列任务，中断在执行的任务；
④TERMINATE: 类似STOP,所有的线程都会停止。
• 线程数量控制参数：用于控制何时创建线程，何时不能创建线程和任务逻辑控制。
①corePoolSize: 池中核心线程的数量；
②maximunPoolSize: 池中最大的线程数量；
③poolSize: 当前池中的线程数量；
• 工作者线程和工作队列：工作者线程从工作队列中获取任务并执行，如果设置了超时参数，工作者线程如果没有任务时，会被回收。工作队列使用的都是阻塞BlockingQueue。
• 线程工厂：ThreadFactory，用于创建新线程，可以自定义线程的一些属性，比如名称等，如果在初始化是没有配置，就会使用默认的ThreadFactory。
• 保持活动的时间：keepAliveTime,当线程池中有多于corePoolSize的线程时，多出的线程空闲时间超过keepAliveTime时将会被清理。
• 拒绝策略：当线程池已经关闭或线程数量达到最大数或工作队列已经饱和时，将拒绝执行新的任务， 主要实现了以下4种拒绝策略：
①AbortPolicy：中断策略，默认的策略，拒绝任务并抛出运行时异常RejectedExecutionException；
②CallerRunsPolicy：使用调用者线程执行任务，除非线程池已经shutdown。
③DiscardPolicy：不能执行的任务将删除；
④DiscardOldestPolicy：如果执行程序未关闭，位于队列头的任务被删除，然后重试执行程序（失败则重复此过程）。

以上的线程池关键要素都会在构造方法中初始化，并根据初始化配置的不同，将产生不同特点的线程池。比如
FixedThreadPool,CachedThreadPool,SingleThreadPool等。

2、执行任务

怎么执行提交到池中的任务是线程池的核心，是由execute方法实现的，执行的过程如下：
①判断核心线程是否都在执行任务，如果不是，则创建新线程来执行任务，如果核心线程都在执行任务，则进入②；
②判断工作队列是否已经满，如果不满，则把任务加入工作队列，如果已经满，则进入步骤③；
③判断线程池的线程是否都处于工作状态，如果没有，则创建一个新线程来执行任务，否则执行拒绝策略。

核心的execute代码如下：

public void execute(Runnable command) {
       if (command == null)
           throw new NullPointerException();
       //如果核心线程数小于当前池中的线程数，则创建新线程执行任务
       if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
           //如果当前池中的线程数大于等于核心线程数，或者创建线程失败，则加入工作队列
           if (runState == RUNNING && workQueue.offer(command)) {
               //如果线程池不再运行或为第一个核心线程执行时
               if (runState != RUNNING || poolSize == 0)
                   ensureQueuedTaskHandled(command);
           }
           //如果线程池不在运行状态或者加入工作队列失败
           //则尝试在最大线程数下创建新线程执行任务
           else if (!addIfUnderMaximumPoolSize(command))
               //再失败，则拒绝任务
               reject(command); // is shutdown or saturated
       }
   }

从代码看出，可以分为以下情况：
①当前的线程数小于核心线程数，则可以继续创建新线程来执行任务；
②如果运行的线程多于corePoolSize，则把任务加入workQueue队列；
③如果无法将任务加入队列（队列已满），则创建新线程执行任务；
④如果创建新线程后的线程数大于maximunPoolSize，则创建失败，并通过拒绝策略拒绝执行新任务；

把任务交给worker线程处理，addThread方法实现了这个逻辑：

private Thread addThread(Runnable firstTask) {
    //构造一个worker任务
    Worker w = new Worker(firstTask);
    //产生一个新线程,把woker封装进去
    Thread t = threadFactory.newThread(w);
    if (t != null) {
        w.thread = t;
        workers.add(w);
        int nt = ++poolSize;
        if (nt > largestPoolSize)
            largestPoolSize = nt;
    }
    return t;
}

3、worker线程实现

worker线程是真正实现执行任务的线程，是通过内部类Worker来实现的，Worker通过在run方法中循环从

workQueue队列中获取任务来执行：

public void run() {
           try {
               Runnable task = firstTask;
               firstTask = null;
               //从阻塞队列中获取任务执行
               while (task != null || (task = getTask()) != null) {
                   runTask(task);
                   task = null;
               }
           } finally {
               workerDone(this);
           }
     }

run方法通过getTask()方法不断从阻塞队列中获取任务，getTask实现过程如下：

Runnable getTask() {
    for (;;) {
        try {
            int state = runState;
            //线程池不在运行状态
            if (state > SHUTDOWN)
                return null;
            Runnable r;
            //关闭状态，把任务poll出
            if (state == SHUTDOWN)  // Help drain queue
                r = workQueue.poll();
            else if (poolSize > corePoolSize || allowCoreThreadTimeOut)
                //超时版本的poll
                r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
            else
                //带有阻塞的take,如果队列为空则阻塞
                r = workQueue.take();
            if (r != null)
                return r;
            //如果worker线程能够退出
            if (workerCanExit()) {
                if (runState >= SHUTDOWN) // Wake up others
                    //清理空闲的worker
                    interruptIdleWorkers();
                return null;
            }
            // Else retry
        } catch (InterruptedException ie) {
            // On interruption, re-check runState
        }
    }
}

Worker类中的runTask方法是实际执行任务的方法，其实现过程如下：

private void runTask(Runnable task) {
        final ReentrantLock runLock = this.runLock;
        //加锁
        runLock.lock();
        try {
            //检查线程池状态和线程状态
            if (runState < STOP &&
                Thread.interrupted() &&
                runState >= STOP)
                thread.interrupt();
            boolean ran = false;
            //执行之前可以插入的操作
            beforeExecute(thread, task);
            try {
                task.run();
                ran = true;
                //执行之后
                afterExecute(task, null);
                ++completedTasks;
            } catch (RuntimeException ex) {
                if (!ran)
                    afterExecute(task, ex);
                throw ex;
            }
        } finally {
            runLock.unlock();
        }
 }

可以看到，在任务执行的前后都可以插入一些代码，由子类来实现监控或其它功能。run方法退出前会执行workDown方法来维护线程池的一些状态信息：

void workerDone(Worker w) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        //增加完成的任务计数
        completedTaskCount += w.completedTasks;
        //从worker队列移出
        workers.remove(w);
        //如果当前的线程数为0，则尝试停止线程池
        if (--poolSize == 0)
            tryTerminate();
    } finally {
        mainLock.unlock();
    }
}

至此，线程池执行任务的相关逻辑结束。

4、关闭线程池

关闭线程池可以通过shutdown()方法或shutdownNow方法来实现，在关闭前都需要进行安全检查，以防恶意代码执行。两个方法不同的地方在于中断worker线程不一样，shutdown调用的是interruptIfIdle方法，而shutdownNow比较粗暴，直接interrupt线程。但shutdownNow会把当前队列中的任务拿出来返回。

public void shutdown() { 
    
SecurityManager security = System.getSecurityManager();
//检查是否有权限
if (security != null)
        security.checkPermission(shutdownPerm);
 
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        //再次检查调用者是否有权限修改线程
        if (security != null) { // Check if caller can modify our threads
            for (Worker w : workers)
                security.checkAccess(w.thread);
        }
        //修改线程池状态
        int state = runState;
        if (state < SHUTDOWN)
            runState = SHUTDOWN;
 
        try {
            //中断所有空闲worker线程
            for (Worker w : workers) {
                w.interruptIfIdle();
            }
        } catch (SecurityException se) { // Try to back out
            runState = state;
            // tryTerminate() here would be a no-op
            throw se;
        }
        //关闭线程池
        tryTerminate(); // Terminate now if pool and queue empty
    } finally {
        mainLock.unlock();
    }
}

void interruptIfIdle() {
        final ReentrantLock runLock = this.runLock;
        //尝试加锁
        if (runLock.tryLock()) {
            try {
                //如果worker中的线程不是当前调用线程
                if (thread != Thread.currentThread())
                    thread.interrupt();
            } finally {
                runLock.unlock();
            }
        }
}

5、并发控制

ThreadPoolExecutor在实现中使用到了ReentrantLock来进行并发控制。核心的表示线程数量的int变量都使用了volatile来保证它的可见性。在Worker实现中也用到了ReentrantLock来控制任务的执行。ThreadPoolExecutor的主要方法都是在lock的控制下进行，比如创建线程时，关闭线程池时。所以，为了避免锁带来的问题，在execute方法中，有这样一个判断：如果当前线程数大于核心线程时，将会把任务加入队列中，这个过程是不需要加锁的，这样就提升了线程池的性能。当线程池预热创建线程数达到核心线程数后都将执行此逻辑。

二、ThreadPoolExecutor的使用

1、创建线程池

一般不推荐直接通过构造函数来创建，应该通过Executors类的工厂方法来创建线程池。因为如果配置的参数不当，线程池的性能可能会受到很大影响。
创建线程池需要指定如下参数：
• corePoolSize: 核心线程数；可以通过prestartAllCoreThreads方法来预热创建核心线程；
• runnableTaskQueue: 任务队列，是阻塞队列blockingQueue的各种实现类。
• maximunPoolSize:池中允许的最大线程数。
• threadFactory: 线程工厂类
• RejectExecutionHandler:拒绝策略处理类。

2、提交任务

提交任务可以使用execute方法或者submit方法，后者会返回一个异步计算结果。

3、关闭线程池

可以通过调用shutdown或shutdownNow方法来关闭线程池。

4、根据任务来配置线程池

任务性质：CPU密集型、IO密集型、混合型
任务的优先级：高、中、低
任务的执行时间：长、中、短
任务的依赖性：是否依赖外部资源，如数据库。

CPU密集型任务可以配置少些线程，IO密集型可以配置多些线程，优先级不同可以用PriorityBlockingQueue来作为工作队列。对于依赖外部资源的任务，因为空闲线程可能较多，可以配置多些线程来利用CPU资源。

ThreadPoolExecutor是一个ExecutorService,实现了Executor接口,是一个线程池服务的实现。通常通过Executors类的工厂方法来创建。

一、实现原理

1、关键要素

要理解线程池的实现原理，主要理解以下的关键要素：

• 线程池的状态：提供了线程池生命周期中所处的不同状态，分别为以下状态(分别用0，1，2，3表示)：
①RUNNING: 可以接收新任务和可以处理队列中的任务；
②SHUTDOWN: 不接收新任务，可以继续处理队列中的任务；
③STOP: 不接收新任务，也不处理队列任务，中断在执行的任务；
④TERMINATE: 类似STOP,所有的线程都会停止。
• 线程数量控制参数：用于控制何时创建线程，何时不能创建线程和任务逻辑控制。
①corePoolSize: 池中核心线程的数量；
②maximunPoolSize: 池中最大的线程数量；
③poolSize: 当前池中的线程数量；
• 工作者线程和工作队列：工作者线程从工作队列中获取任务并执行，如果设置了超时参数，工作者线程如果没有任务时，会被回收。工作队列使用的都是阻塞BlockingQueue。
• 线程工厂：ThreadFactory，用于创建新线程，可以自定义线程的一些属性，比如名称等，如果在初始化是没有配置，就会使用默认的ThreadFactory。
• 保持活动的时间：keepAliveTime,当线程池中有多于corePoolSize的线程时，多出的线程空闲时间超过keepAliveTime时将会被清理。
• 拒绝策略：当线程池已经关闭或线程数量达到最大数或工作队列已经饱和时，将拒绝执行新的任务， 主要实现了以下4种拒绝策略：
①AbortPolicy：中断策略，默认的策略，拒绝任务并抛出运行时异常RejectedExecutionException；
②CallerRunsPolicy：使用调用者线程执行任务，除非线程池已经shutdown。
③DiscardPolicy：不能执行的任务将删除；
④DiscardOldestPolicy：如果执行程序未关闭，位于队列头的任务被删除，然后重试执行程序（失败则重复此过程）。

2、执行任务

核心的execute代码如下：

public void execute(Runnable command) {
       if (command == null)
           throw new NullPointerException();
       //如果核心线程数小于当前池中的线程数，则创建新线程执行任务
       if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
           //如果当前池中的线程数大于等于核心线程数，或者创建线程失败，则加入工作队列
           if (runState == RUNNING && workQueue.offer(command)) {
               //如果线程池不再运行或为第一个核心线程执行时
               if (runState != RUNNING || poolSize == 0)
                   ensureQueuedTaskHandled(command);
           }
           //如果线程池不在运行状态或者加入工作队列失败
           //则尝试在最大线程数下创建新线程执行任务
           else if (!addIfUnderMaximumPoolSize(command))
               //再失败，则拒绝任务
               reject(command); // is shutdown or saturated
       }
   }

把任务交给worker线程处理，addThread方法实现了这个逻辑：

private Thread addThread(Runnable firstTask) {
    //构造一个worker任务
    Worker w = new Worker(firstTask);
    //产生一个新线程,把woker封装进去
    Thread t = threadFactory.newThread(w);
    if (t != null) {
        w.thread = t;
        workers.add(w);
        int nt = ++poolSize;
        if (nt > largestPoolSize)
            largestPoolSize = nt;
    }
    return t;
}

3、worker线程实现

worker线程是真正实现执行任务的线程，是通过内部类Worker来实现的，Worker通过在run方法中循环从

workQueue队列中获取任务来执行：

public void run() {
           try {
               Runnable task = firstTask;
               firstTask = null;
               //从阻塞队列中获取任务执行
               while (task != null || (task = getTask()) != null) {
                   runTask(task);
                   task = null;
               }
           } finally {
               workerDone(this);
           }
     }

run方法通过getTask()方法不断从阻塞队列中获取任务，getTask实现过程如下：

Runnable getTask() {
    for (;;) {
        try {
            int state = runState;
            //线程池不在运行状态
            if (state > SHUTDOWN)
                return null;
            Runnable r;
            //关闭状态，把任务poll出
            if (state == SHUTDOWN)  // Help drain queue
                r = workQueue.poll();
            else if (poolSize > corePoolSize || allowCoreThreadTimeOut)
                //超时版本的poll
                r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
            else
                //带有阻塞的take,如果队列为空则阻塞
                r = workQueue.take();
            if (r != null)
                return r;
            //如果worker线程能够退出
            if (workerCanExit()) {
                if (runState >= SHUTDOWN) // Wake up others
                    //清理空闲的worker
                    interruptIdleWorkers();
                return null;
            }
            // Else retry
        } catch (InterruptedException ie) {
            // On interruption, re-check runState
        }
    }
}

Worker类中的runTask方法是实际执行任务的方法，其实现过程如下：

private void runTask(Runnable task) {
        final ReentrantLock runLock = this.runLock;
        //加锁
        runLock.lock();
        try {
            //检查线程池状态和线程状态
            if (runState < STOP &&
                Thread.interrupted() &&
                runState >= STOP)
                thread.interrupt();
            boolean ran = false;
            //执行之前可以插入的操作
            beforeExecute(thread, task);
            try {
                task.run();
                ran = true;
                //执行之后
                afterExecute(task, null);
                ++completedTasks;
            } catch (RuntimeException ex) {
                if (!ran)
                    afterExecute(task, ex);
                throw ex;
            }
        } finally {
            runLock.unlock();
        }
 }

可以看到，在任务执行的前后都可以插入一些代码，由子类来实现监控或其它功能。run方法退出前会执行workDown方法来维护线程池的一些状态信息：

void workerDone(Worker w) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        //增加完成的任务计数
        completedTaskCount += w.completedTasks;
        //从worker队列移出
        workers.remove(w);
        //如果当前的线程数为0，则尝试停止线程池
        if (--poolSize == 0)
            tryTerminate();
    } finally {
        mainLock.unlock();
    }
}

至此，线程池执行任务的相关逻辑结束。

4、关闭线程池

public void shutdown() { 
    
SecurityManager security = System.getSecurityManager();
//检查是否有权限
if (security != null)
        security.checkPermission(shutdownPerm);
 
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        //再次检查调用者是否有权限修改线程
        if (security != null) { // Check if caller can modify our threads
            for (Worker w : workers)
                security.checkAccess(w.thread);
        }
        //修改线程池状态
        int state = runState;
        if (state < SHUTDOWN)
            runState = SHUTDOWN;
 
        try {
            //中断所有空闲worker线程
            for (Worker w : workers) {
                w.interruptIfIdle();
            }
        } catch (SecurityException se) { // Try to back out
            runState = state;
            // tryTerminate() here would be a no-op
            throw se;
        }
        //关闭线程池
        tryTerminate(); // Terminate now if pool and queue empty
    } finally {
        mainLock.unlock();
    }
}

void interruptIfIdle() {
        final ReentrantLock runLock = this.runLock;
        //尝试加锁
        if (runLock.tryLock()) {
            try {
                //如果worker中的线程不是当前调用线程
                if (thread != Thread.currentThread())
                    thread.interrupt();
            } finally {
                runLock.unlock();
            }
        }
}

5、并发控制

二、ThreadPoolExecutor的使用

1、创建线程池

一般不推荐直接通过构造函数来创建，应该通过Executors类的工厂方法来创建线程池。因为如果配置的参数不当，线程池的性能可能会受到很大影响。
创建线程池需要指定如下参数：
• corePoolSize: 核心线程数；可以通过prestartAllCoreThreads方法来预热创建核心线程；
• runnableTaskQueue: 任务队列，是阻塞队列blockingQueue的各种实现类。
• maximunPoolSize:池中允许的最大线程数。
• threadFactory: 线程工厂类
• RejectExecutionHandler:拒绝策略处理类。

2、提交任务

提交任务可以使用execute方法或者submit方法，后者会返回一个异步计算结果。

3、关闭线程池

可以通过调用shutdown或shutdownNow方法来关闭线程池。

4、根据任务来配置线程池

任务性质：CPU密集型、IO密集型、混合型
任务的优先级：高、中、低
任务的执行时间：长、中、短
任务的依赖性：是否依赖外部资源，如数据库。

应用商城

合作伙伴

开发者

支持与服务

了解天翼云

ThreadPoolExecutor实现分析

一、实现原理

1、关键要素

2、执行任务

3、worker线程实现

4、关闭线程池

5、并发控制

二、ThreadPoolExecutor的使用

1、创建线程池

2、提交任务

3、关闭线程池

4、根据任务来配置线程池

ThreadPoolExecutor实现分析

一、实现原理

1、关键要素

2、执行任务

3、worker线程实现

4、关闭线程池

5、并发控制

二、ThreadPoolExecutor的使用

1、创建线程池

2、提交任务

3、关闭线程池

4、根据任务来配置线程池