文章目录

0、Executor 框架中类的继承关系一、ThreadPoolExecutor1.1、概述1.2、核心构造方法讲解1.3、execute(Runable)方法执行过程 二、Executors 提供的线程池配置方案2.1、Executors 内置方法： 三、ThreadPoolExecutor 与 Executors 区别相同点：Executors 的不同点：ThreadPoolExecutor 的不同点： 四、自定义 非阻塞线程池五、自定义 阻塞线程池

0、Executor 框架中类的继承关系

/** <pre> 一、线程池的体系结构： Executor 负责线程的使用和调度的 根接口 |--ExecutorService 子接口： 线程池的主要接口 |--ThreadPoolExecutor 线程池的实现类 |--ScheduledExceutorService 子接口： 负责线程的调度 |--ScheduledThreadPoolExecutor : 继承ThreadPoolExecutor，实现了ScheduledExecutorService 二、工具类 ： Executors ExecutorService pool = Executors.newFixedThreadPool() ： 创建固定大小的线程池 ExecutorService pool = Executors.newCachedThreadPool() ： 缓存线程池，线程池的数量不固定，可以根据需求自动的更改数量。 ExecutorService pool = Executors.newSingleThreadExecutor() ：创建单个线程池。 线程池中只有一个线程 ScheduledExecutorService pool = Executors.newScheduledThreadPool() ： 创建固定大小的线程，可以延迟或定时的执行任务 </pre> */

一、ThreadPoolExecutor

ThreadPoolExecutor 是 java提供的一个线程池工具类

1.1、概述

1、ThreadPoolExecutor 作为 java.util.concurrent 包对外提供基础实现，以内部线程池的形式对外提供管理任务执行，线程调度，线程池管理等等服务； 2、Executors 方法提供的线程服务，都是通过ThreadPoolExecutor 设置不同参数来实现不同的线程池机制。 3、先来了解其线程池管理的机制，有助于正确使用，避免错误使用导致严重故障。同时可以根据自己的需求实现自己的线程池

1.2、核心构造方法讲解

ThreadPoolExecutor 最核心的构造方法 ：

public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, defaultHandler); }

构造方法参数讲解 参数名 -------------- 作用

1、corePoolSize 核心线程池大小

2、maximumPoolSize 最大线程池大小

3、keepAliveTime 线程池中超过corePoolSize数目的空闲线程最大存活时间；可以allowCoreThreadTimeOut(true) 使得核心线程有效时间

3、TimeUnit unit 时间单位

4、workQueue 阻塞任务队列

5、threadFactory 新建线程工厂

6、RejectedExecutionHandler 当提交任务数超过maxmumPoolSize+workQueue之和时，任务会交给 RejectedExecutionHandler来处理

重点讲解 其中比较容易让人误解的是：corePoolSize，maximumPoolSize，workQueue之间关系。

当 线程池 小于 corePoolSize 时，新任务 放到 线程池 中去执行；

当 线程池 达到 corePoolSize 时，新任务 放到 workQueue 中，等待线程池中任务调度执行 ；

当 workQueue 已满，且 maximumPoolSize > corePoolSize 时，新任务 会创建 新线程 去执行任务；

当 任务的数量 超过 maximumPoolSize 时，新任务 由 RejectedExecutionHandler 处理

当 线程池 中超过 corePoolSize 线程，空闲时间达到 keepAliveTime 时，关闭空闲线程

当设置allowCoreThreadTimeOut(true)时，线程池中corePoolSize线程空闲时间达到keepAliveTime也将关闭 。

1.3、execute(Runable)方法执行过程

如果此时线程池中的数量小于 corePoolSize，即使线程池中的线程都处于空闲状态，也要创建新的线程来处理被添加的任务。

如果此时线程池中的数量 等于 corePoolSize，但是缓冲队列 workQueue 未满，那么任务被放入缓冲队列。

如果此时线程池中的数量大于 corePoolSize ，缓冲队列 workQueue 满，并且线程池中的数量小于 maximumPoolSize ，建新的线程来处理被添加的任务。

如果此时线程池中的数量大于 corePoolSize ，缓冲队列 workQueue 满，并且线程池中的数量等于 maximumPoolSize，那么通过 handler 所指定的策略来处理此任务。也就是：处理任务的优先级为：核心线程 corePoolSize、任务队列 workQueue、最大线程 maximumPoolSize ，如果三者都满了，使用 handler 处理被拒绝的任务。

当线程池中的线程数量大于 corePoolSize 时，如果某线程空闲时间超过keepAliveTime ，线程将被终止。这样，线程池可以动态的调整池中的线程数。

二、Executors 提供的线程池配置方案

Executors 是一个工具类，类似于 Collections 、Arrays 等 。提供工厂方法来创建不同类型的线程池，比如 FixedThreadPool 或 CachedThreadPool 。

Executors 内置方法 使用的是 ThreadPoolExecutor 类 来定义的 。

2.1、Executors 内置方法：

// 创建固定大小的线程池 ExecutorService pool = Executors.newFixedThreadPool() ; // 缓存线程池，线程池的数量不固定，可以根据需求自动的更改数量 ExecutorService pool = Executors.newCachedThreadPool(); // 创建单个线程池。 线程池中只有一个线程 ExecutorService pool = Executors.newSingleThreadExecutor() ; // 创建固定大小的线程，可以延迟或定时的执行任务 ScheduledExecutorService pool = Executors.newScheduledThreadPool();

三、ThreadPoolExecutor 与 Executors 区别

相同点：

都可以创建 线程池 ；

Executors 的不同点：

Executors 是一个工具类，内置多个创建 线程池 相关的静态方法； Executors 中线程池的方法使用的 ThreadPoolExecutor 创建的；

ThreadPoolExecutor 的不同点：

ThreadPoolExecutor 是一个Executor 接口的实现类 。

四、自定义 非阻塞线程池

import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.RejectedExecutionHandler; import java.util.concurrent.ThreadFactory; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; public class CustomThreadPoolExecutor { private ThreadPoolExecutor pool = null; /** * 线程池初始化方法 * * corePoolSize 核心线程池大小----10 * maximumPoolSize 最大线程池大小----30 * keepAliveTime 线程池中超过corePoolSize数目的空闲线程最大存活时间----30 * TimeUnit keepAliveTime时间单位----TimeUnit.MINUTES * workQueue 阻塞队列----new ArrayBlockingQueue<Runnable>(10)====10容量的阻塞队列 * threadFactory 新建线程工厂----new CustomThreadFactory()====定制的线程工厂 * rejectedExecutionHandler 当提交任务数超过maxmumPoolSize+workQueue之和时, * 即当提交第41个任务时(前面线程都没有执行完,此测试方法中用sleep(100)), * 任务会交给RejectedExecutionHandler来处理 */ public void init() { pool = new ThreadPoolExecutor( 10, 30, 30, TimeUnit.MINUTES, new ArrayBlockingQueue<Runnable>(10), new CustomThreadFactory(), new CustomRejectedExecutionHandler()); } public void destory() { if(pool != null) { pool.shutdownNow(); } } public ExecutorService getCustomThreadPoolExecutor() { return this.pool; } private class CustomThreadFactory implements ThreadFactory { private AtomicInteger count = new AtomicInteger(0); @Override public Thread newThread(Runnable r) { Thread t = new Thread(r); String threadName = CustomThreadPoolExecutor.class.getSimpleName() + count.addAndGet(1); System.out.println(threadName); t.setName(threadName); return t; } } private class CustomRejectedExecutionHandler implements RejectedExecutionHandler { @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { // 记录异常 // 报警处理等 System.out.println("error............."); } } // 测试构造的线程池 public static void main(String[] args) { CustomThreadPoolExecutor exec = new CustomThreadPoolExecutor(); // 1.初始化 exec.init(); ExecutorService pool = exec.getCustomThreadPoolExecutor(); for(int i=1; i<100; i++) { System.out.println("提交第" + i + "个任务!"); pool.execute(new Runnable() { @Override public void run() { try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("running====="); } }); } // 2.销毁----此处不能销毁,因为任务没有提交执行完,如果销毁线程池,任务也就无法执行了 // exec.destory(); try { Thread.sleep(10000); } catch (InterruptedException e) { e.printStackTrace(); } } }

方法中建立一个核心线程数为30个，缓冲队列有10个的线程池。每个线程任务，执行时会先睡眠3秒，保证提交10任务时，线程数目被占用完，再提交30任务时，阻塞队列被占用完，，这样提交第41个任务是，会交给CustomRejectedExecutionHandler 异常处理类来处理。

提交任务的代码如下：

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); }

注意：41以后提交的任务就不能正常处理了，因为，execute中提交到任务队列是用的offer方法，如上面代码，这个方法是非阻塞的，所以就会交给CustomRejectedExecutionHandler 来处理，所以对于大数据量的任务来说，这种线程池，如果不设置队列长度会OOM，设置队列长度，会有任务得不到处理，接下来我们构建一个阻塞的自定义线程池

五、自定义 阻塞线程池

package com.tongbanjie.trade.test.commons; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.RejectedExecutionHandler; import java.util.concurrent.ThreadFactory; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; public class CustomThreadPoolExecutor { private ThreadPoolExecutor pool = null; /** * 线程池初始化方法 * * corePoolSize 核心线程池大小----1 * maximumPoolSize 最大线程池大小----3 * keepAliveTime 线程池中超过corePoolSize数目的空闲线程最大存活时间----30+单位TimeUnit * TimeUnit keepAliveTime时间单位----TimeUnit.MINUTES * workQueue 阻塞队列----new ArrayBlockingQueue<Runnable>(5)====5容量的阻塞队列 * threadFactory 新建线程工厂----new CustomThreadFactory()====定制的线程工厂 * rejectedExecutionHandler 当提交任务数超过maxmumPoolSize+workQueue之和时, * 即当提交第41个任务时(前面线程都没有执行完,此测试方法中用sleep(100)), * 任务会交给RejectedExecutionHandler来处理 */ public void init() { pool = new ThreadPoolExecutor( 1, 3, 30, TimeUnit.MINUTES, new ArrayBlockingQueue<Runnable>(5), new CustomThreadFactory(), new CustomRejectedExecutionHandler()); } public void destory() { if(pool != null) { pool.shutdownNow(); } } public ExecutorService getCustomThreadPoolExecutor() { return this.pool; } private class CustomThreadFactory implements ThreadFactory { private AtomicInteger count = new AtomicInteger(0); @Override public Thread newThread(Runnable r) { Thread t = new Thread(r); String threadName = CustomThreadPoolExecutor.class.getSimpleName() + count.addAndGet(1); System.out.println(threadName); t.setName(threadName); return t; } } private class CustomRejectedExecutionHandler implements RejectedExecutionHandler { @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { try { // 核心改造点，由blockingqueue的offer改成put阻塞方法 executor.getQueue().put(r); } catch (InterruptedException e) { e.printStackTrace(); } } } // 测试构造的线程池 public static void main(String[] args) { CustomThreadPoolExecutor exec = new CustomThreadPoolExecutor(); // 1.初始化 exec.init(); ExecutorService pool = exec.getCustomThreadPoolExecutor(); for(int i=1; i<100; i++) { System.out.println("提交第" + i + "个任务!"); pool.execute(new Runnable() { @Override public void run() { try { System.out.println(">>>task is running====="); TimeUnit.SECONDS.sleep(10); } catch (InterruptedException e) { e.printStackTrace(); } } }); } // 2.销毁----此处不能销毁,因为任务没有提交执行完,如果销毁线程池,任务也就无法执行了 // exec.destory(); try { Thread.sleep(10000); } catch (InterruptedException e) { e.printStackTrace(); } } }

解释：当提交任务被拒绝时，进入拒绝机制，我们实现拒绝方法，把任务重新用阻塞提交方法put提交，实现阻塞提交任务功能，防止队列过大，OOM，提交被拒绝方法在下面

public void execute(Runnable command) { if (command == null) throw new NullPointerException(); 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)) // 进入拒绝机制， 我们把runnable任务拿出来，重新用阻塞操作put，来实现提交阻塞功能 reject(command); }

总结： 1、用ThreadPoolExecutor自定义线程池，看线程是的用途，如果任务量不大，可以用无界队列，如果任- - 务量非常大，要用有界队列，防止OOM 2、如果任务量很大，还要求每个任务都处理成功，要对提交的任务进行阻塞提交，重写拒绝机制，改为阻塞提交。保证不抛弃一个任务 3、最大线程数一般设为2N+1最好，N是CPU核数

4、核心线程数，看应用，如果是任务，一天跑一次，设置为0，合适，因为跑完就停掉了，如果是常用线程池，看任务量，是保留一个核心还是几个核心线程数 5、如果要获取任务执行结果，用CompletionService，但是注意，获取任务的结果的要重新开一个线程获取，如果在主线程获取，就要等任务都提交后才获取，就会阻塞大量任务结果，队列过大OOM，所以最好异步开个线程获取结果