（一）执行器 Executor
执行器内部有一个任务队列 BlockingQueue<Runnable> 和一个线程池 Workers，客户端不断地往任务队列中添加任务，工作者线程则不断地从任务队列移除并执行任务
（二）执行器的创建
//取自ThreadPoolExecutor构造方法
/**
* @param corePoolSize 最少线程数
* @param maximumPoolSize 最多线程数
* @param keepAliveTime 线程多久不执行任务就会被关闭
* @param unit keepAliveTime 的时间单位
* @param workQueue 任务队列
* @param threadFactory 线程工厂
* @param handler 饱和策略
*/
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
//...
}
1） 线程工厂
默认ThreadFactory的实现类为Executors.DefaultThreadFactory，但在实际使用中，不能使用默认的线程工厂，因为必须要为线程指定有意义的线程名
取自Executors$DefaultThreadFactory
static class DefaultThreadFactory implements ThreadFactory {
// 线程池计数
private static final AtomicInteger poolNumber = new AtomicInteger(1);
private final ThreadGroup group;
// 池中的线程计数
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final String namePrefix;
DefaultThreadFactory() {
SecurityManager s = System.getSecurityManager();
group = (s != null) ? s.getThreadGroup() :
Thread.currentThread().getThreadGroup();
namePrefix = "pool-" +
poolNumber.getAndIncrement() +
"-thread-";
}
public Thread newThread(Runnable r) {
Thread t = new Thread(group, r,
namePrefix + threadNumber.getAndIncrement(),
0);
if (t.isDaemon())
t.setDaemon(false);
if (t.getPriority() != Thread.NORM_PRIORITY)
t.setPriority(Thread.NORM_PRIORITY);
return t;
}
}
//自定义线程工厂-命名线程
ThreadFactory threadFactory = new ThreadFactory() {
private final AtomicInteger counter = new AtomicInteger(1);
public Thread newThread(Runnable r) {
return new Thread(r, "hippo_monitor_" + counter.getAndIncrement());
}
}
2） 饱和策略
饱和策略是在任务提交时可能发生
默认采用AbortPolicy作为饱和策略
1）中断提交者策略
public static class AbortPolicy implements RejectedExecutionHandler {
public AbortPolicy() { }
// 提交任务时抛出RejectedExecutionException异常
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
}
2）丢弃任务策略
public static class DiscardPolicy implements RejectedExecutionHandler {
public DiscardPolicy() { }
// 提交任务时直接丢弃
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}
3）提交者自己运行策略
public static class CallerRunsPolicy implements RejectedExecutionHandler {
public CallerRunsPolicy() { }
// 提交任务时由提交者自己运行
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}
3） Executors创建的4大执行器的缺点
1）CachedThreadPool、ScheduledThreadPool 允许创建的线程数量为Integer.MAX_VALUE，为每一个任务分配一个线程，可能会创建大量的线程，从而导致资源耗尽
2）FixedThreadPool、SingleThreadPool 使用LinkedBlockingQueue作为任务队列,队列的最大长度为Integer.MAX_VALUE，可能会堆积大量的任务，从而导致资源耗尽
3）以上四种执行器都使用默认的线程工厂，线程名都是：pool-数字-thread-数字，不方便在出错时回溯
由上所述，规范下，线程池不允许使用Executors创建，而是直接通过调用ThreadPoolExecutor的构造方法创建
//Executors.newCachedThreadPool()
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
//Executors.newScheduledThreadPool(10)
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue());
}
//Executors.newSingleThreadExecutor()
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
//Executors.newFixedThreadPool(10)
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
（三）生命周期
// 当前的运行状态
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 所有的状态
private static final int RUNNING = -1 << COUNT_BITS;
private static final int SHUTDOWN = 0 << COUNT_BITS;
private static final int STOP = 1 << COUNT_BITS;
private static final int TIDYING = 2 << COUNT_BITS;
private static final int TERMINATED = 3 << COUNT_BITS;
一些生命周期的方法
1)
private static boolean isRunning(int c) {
return c < SHUTDOWN;
}
2)
public boolean isShutdown() {
return ! isRunning(ctl.get());
}
3)
public boolean isTerminated() {
return runStateAtLeast(ctl.get(), TERMINATED);
}
4)
public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 轮询判断运行状态是否为TERMINATED以及是否到点
for (;;) {
if (runStateAtLeast(ctl.get(), TERMINATED))
return true;
if (nanos <= 0)
return false;
nanos = termination.awaitNanos(nanos);
}
} finally {
mainLock.unlock();
}
}
5)
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(SHUTDOWN);
// 中断所有的空闲线程
interruptIdleWorkers();
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
tryTerminate();
}
6)
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(STOP);
// 中断所有的线程
interruptWorkers();
tasks = drainQueue();
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}
网友评论