java线程池之ThreadPoolExecutor
2017-10-10 本文已影响19人
java面试收割机
一、ThreadPoolExecutor类
核心构造方法：
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
if (corePoolSize < 0 ||
maximumPoolSize <= 0 ||
maximumPoolSize < corePoolSize ||
keepAliveTime < 0)
throw new IllegalArgumentException();
if (workQueue == null || threadFactory == null || handler == null)
throw new NullPointerException();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
变量参数含义：
int corePoolSize; // 线程池维护线程的最小数量，哪怕是空闲的
int maximumPoolSize; // 线程池维护的最大线程数量
long keepAliveTime; // 线程池维护线程所允许的空闲时间
TimeUnit unit //keepAliveTime时间单位
BlockingQueue<Runnable> workQueue; // 阻塞队列
ThreadFactory threadFactory; // ThreadFactory对象，用于创建线程。
RejectedExecutionHandler handler;// 拒绝策略的处理句柄
重要参数corePoolSize，maximumPoolSize，workQueue之间关系：
1、当运行的线程少于 corePoolSize，则创建新线程来处理请求，即使其他线程是空闲的。
2、如果corePoolSize 和 maximumPoolSize相同，如果运行的线程与corePoolSize相同，当有新请求过来时，若workQueue未满，则将请求放入workQueue队列中，等待有空闲的线程去从workQueue中取任务
3、如果运行的线程多于 corePoolSize 而少于 maximumPoolSize，则仅当队列满时才创建新线程才创建新的线程去处理请求；
4、如果运行的线程多于corePoolSize 并且等于maximumPoolSize，若队列已经满了，则通过handler所指定的策略来处理新请求；
如果将 maximumPoolSize 设置为基本的无界值（如 Integer.MAX_VALUE），则允许池适应任意数量的并发任务
如果三者都满了，使用handler处理被拒绝的任务。当池中的线程数大于corePoolSize的时候，多余的空闲线程会等待keepAliveTime长的时间自行销毁。当设置allowCoreThreadTimeOut(true)时，线程池中corePoolSize线程空闲时间达到keepAliveTime也将关闭
二、线程池拒绝策略
当corePoolSize、workQueue、maximumPoolSize都满了，还有新任务进来，线程池将使用拒绝策略处理线程，常用四种拒绝策略如下：
ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
ThreadPoolExecutor.DiscardOldestPolicy：如果执行程序尚未关闭，则位于工作队列头部的任务将被删除，然后重试执行程序
ThreadPoolExecutor.CallerRunsPolicy：线程调用运行该任务的 execute 本身。此策略提供简单的反馈控制机制，能够减缓新任务的提交速度
测试线程代码：
class MyRunnable implements Runnable {
private String name;
public MyRunnable(String name) {
this.name = name;
}
@Override
public void run() {
try {
System.out.println(this.name + " is running.");
Thread.sleep(100);
} catch (Exception e) {
e.printStackTrace();
}
}
}
AbortPolicy策略：丢弃任务并抛出RejectedExecutionException异常
public class AbortPolicyDemo {
private static final int THREADS_SIZE = 1;
private static final int CAPACITY = 1;
public static void main(String[] args) throws Exception {
// 创建线程池。线程池的"最大池大小"和"核心池大小"都为1(THREADS_SIZE)，"线程池"的阻塞队列容量为1(CAPACITY)。
ThreadPoolExecutor pool = new ThreadPoolExecutor(THREADS_SIZE, THREADS_SIZE, 0, TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(CAPACITY));
// 设置线程池的拒绝策略为"抛出异常"
pool.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
try {
// 新建10个任务，并将它们添加到线程池中。
for (int i = 0; i < 10; i++) {
Runnable myrun = new MyRunnable("task-"+i);
pool.execute(myrun);
}
} catch (RejectedExecutionException e) {
e.printStackTrace();
// 关闭线程池
pool.shutdown();
}
}
}
结果：后面的任务将被丢弃，并抛出异常
java.util.concurrent.RejectedExecutionException
at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:1774)
at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:768)
at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:656)
at AbortPolicyDemo.main(AbortPolicyDemo.java:27)
task-0 is running.
task-1 is running.
DiscardPolicy策略：丢弃任务，但是不抛出异常
public class DiscardPolicyDemo {
private static final int THREADS_SIZE = 1;
private static final int CAPACITY = 1;
public static void main(String[] args) throws Exception {
// 创建线程池。线程池的"最大池大小"和"核心池大小"都为1(THREADS_SIZE)，"线程池"的阻塞队列容量为1(CAPACITY)。
ThreadPoolExecutor pool = new ThreadPoolExecutor(THREADS_SIZE, THREADS_SIZE, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(CAPACITY));
// 设置线程池的拒绝策略为"丢弃"
pool.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());
// 新建10个任务，并将它们添加到线程池中。
for (int i = 0; i < 10; i++) {
Runnable myrun = new MyRunnable("task-"+i);
pool.execute(myrun);
}
// 关闭线程池
pool.shutdown();
}
}
结果：
task-0 is running.
task-1 is running.
DiscardOldestPolicy策略：丢弃队列最前面的任务，然后重新尝试执行任务
public class DiscardOldestPolicyDemo {
private static final int THREADS_SIZE = 1;
private static final int CAPACITY = 1;
public static void main(String[] args) throws Exception {
// 创建线程池。线程池的"最大池大小"和"核心池大小"都为1(THREADS_SIZE)，"线程池"的阻塞队列容量为1(CAPACITY)。
ThreadPoolExecutor pool = new ThreadPoolExecutor(THREADS_SIZE, THREADS_SIZE, 0, TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(CAPACITY));
// 设置线程池的拒绝策略为"DiscardOldestPolicy"
pool.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());
// 新建10个任务，并将它们添加到线程池中。
for (int i = 0; i < 10; i++) {
Runnable myrun = new MyRunnable("task-"+i);
pool.execute(myrun);
}
// 关闭线程池
pool.shutdown();
}
}
结果：丢弃阻塞队列中末尾的任务，然后将被拒绝的任务添加到末尾
task-0 is running.
task-9 is running.
CallerRunsPolicy：线程调用运行该任务的 execute 本身。此策略提供简单的反馈控制机制，能够减缓新任务的提交速度
public class CallerRunsPolicyDemo {
private static final int THREADS_SIZE = 1;
private static final int CAPACITY = 1;
public static void main(String[] args) throws Exception {
// 创建线程池。线程池的"最大池大小"和"核心池大小"都为1(THREADS_SIZE)，"线程池"的阻塞队列容量为1(CAPACITY)。
ThreadPoolExecutor pool = new ThreadPoolExecutor(THREADS_SIZE, THREADS_SIZE, 0, TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(CAPACITY));
// 设置线程池的拒绝策略为"CallerRunsPolicy"
pool.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
// 新建10个任务，并将它们添加到线程池中。
for (int i = 0; i < 10; i++) {
Runnable myrun = new MyRunnable("task-"+i);
pool.execute(myrun);
}
// 关闭线程池
pool.shutdown();
}
}
结果：某一次运行结果，当有新任务添加到线程池被拒绝时，线程池会将被拒绝的任务添加到"线程池正在运行的线程"中去运行
task-2 is running.
task-3 is running.
task-4 is running.
task-5 is running.
task-6 is running.
task-7 is running.
task-8 is running.
task-9 is running.
task-0 is running.
task-1 is running.
