java.util.concurrent学习(七) ThreadPoolExecutor

本文主要是介绍java.util.concurrent学习(七) ThreadPoolExecutor，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

ThreadPoolExecutor继承了AbstractExecutorService，是线程池工厂Executors创建线程池的主要实现方式。通过线程池工厂可以快速创建线程池，然而这种方式也有一定的弊端。例如：Executors.newFixedThreadPool(int nThreads)创建的线程池核心线程数固定，不能灵活地扩展最大线程数；newCachedThreadPool方式将最大线程数设计为Integer.MAX_VALUE，这种方式容易造成内存不足，不推荐使用。如果想通过线程池的方式来实现业务的话，最好对线程池进行系统的了解，设计真正符合项目的线程池才是最好的解决办法。下面我们就来一起学习一下吧。

1.通过最基础的构造方法认识ThreadPoolExecutor

  public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler)

ThreadPoolExecutor有四个构造方法，其他三个都是通过调用这个构造方法来实现参数控制。来看源码：

/***corePoolSize    核心线程数，除非设置了allowCoreThreadTimeOut，否则核心线程将会一直在*                保存在池中，即使是闲置的。核心线程可以帮助我们快速执行任务，避免了创建*                线程带来的消耗。(corePoolSize>=0)**maximumPoolSize 最大线程数，线程池中允许存在的最大线程数，可以用来控制内存使用，避免不*                必要的线程创建。(maximumPoolSize>0&&maximumPoolSize>=corePoolSize)**keepAliveTime   非核心线程闲置超时时间。当线程池的线程数超过核心线程数时，非核心线程将*                等待新任务的时间不能超过keepAliveTime,一旦超过将自动terminating。      *                (keepAliveTime>0)**unit            keepAliveTime的时间单位。**workQueue       等待执行的线程队列，该队列中只保存被submitted的线程。(not null)**threadFactory   生产线程的工厂。(not null)**handler         线程拒绝策列。当线程池和队列已满时，通过该handler来处理新提交的任务。*                (not null)**/
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.acc = System.getSecurityManager() == null ?null :AccessController.getContext();this.corePoolSize = corePoolSize;this.maximumPoolSize = maximumPoolSize;this.workQueue = workQueue;this.keepAliveTime = unit.toNanos(keepAliveTime);this.threadFactory = threadFactory;this.handler = handler;}

2.属性设置 allowCoreThreadTimeOut

  /***该参数需要配合keepAliveTime来使用，当设置为true时，核心线程会像非核心线程一样遵循*keepAliveTime的超时规则。默认false。**/
public void allowCoreThreadTimeOut(boolean value) {if (value && keepAliveTime <= 0)throw new IllegalArgumentException("Core threads must have nonzero keep alive times");if (value != allowCoreThreadTimeOut) {allowCoreThreadTimeOut = value;if (value)//当设置为true时，立即打断限制的核心线程interruptIdleWorkers();}}private void interruptIdleWorkers() {interruptIdleWorkers(false);}//onlyOne  是否只打断一个线程
private void interruptIdleWorkers(boolean onlyOne) {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {for (Worker w : workers) {Thread t = w.thread;if (!t.isInterrupted() && w.tryLock()) {try {t.interrupt();} catch (SecurityException ignore) {} finally {w.unlock();}}if (onlyOne)break;}} finally {mainLock.unlock();}}

3.提交线程execute

  /***@param command 要提交的任务（not null）**@desc  提交一个任务，线程池会在将来的某一时刻执行该任务，如果线程池已满，则会通过拒绝策略来 *       处理该任务**/
public void execute(Runnable command) {if (command == null)throw new NullPointerException();/** 总共有三步:*** 1. 检查当前线程数是否少于核心线程数。如果当前线程数少于核心线程数，*那么创建一个新的线程来执行该任务。通过调用addWorker进行原子操作检查*线程状态和数目，从而避免出现错误的溢出警戒值。** 2. 如果一个任务可以成功的添加到队列，那么仍然需要二次检查是否需要添*加一个线程或者线程池是否已经关闭。** 3. 如果无法增加任务到队列，那就尝试添加一个线程。如果失败了，那么需要*通过拒绝策略来处理。*/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);}private boolean addWorker(Runnable firstTask, boolean core) {retry:for (;;) {int c = ctl.get();int rs = runStateOf(c);// Check if queue empty only if necessary.if (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))return false;for (;;) {int wc = workerCountOf(c);if (wc >= CAPACITY ||wc >= (core ? corePoolSize : maximumPoolSize))return false;if (compareAndIncrementWorkerCount(c))break retry;c = ctl.get();  // Re-read ctlif (runStateOf(c) != rs)continue retry;// else CAS failed due to workerCount change; retry inner loop}}boolean workerStarted = false;boolean workerAdded = false;Worker w = null;try {w = new Worker(firstTask);final Thread t = w.thread;if (t != null) {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {// Recheck while holding lock.// Back out on ThreadFactory failure or if// shut down before lock acquired.int rs = runStateOf(ctl.get());if (rs < SHUTDOWN ||(rs == SHUTDOWN && firstTask == null)) {if (t.isAlive()) // precheck that t is startablethrow new IllegalThreadStateException();workers.add(w);int s = workers.size();if (s > largestPoolSize)largestPoolSize = s;workerAdded = true;}} finally {mainLock.unlock();}if (workerAdded) {t.start();workerStarted = true;}}} finally {if (! workerStarted)addWorkerFailed(w);}return workerStarted;}

 

这篇关于java.util.concurrent学习(七) ThreadPoolExecutor的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---