首页 > 技术文章 > Java线程池的内部实现

junjiang3 2018-03-11 16:39 原文

一、线程池介绍

线程是稀缺资源,如果无限制的创建,不仅会消耗系统资源,还会降低系统的稳定性,合理的使用线程池可以对线程进行统一的分配、调优和监控,并有以下好处:

(1)降低资源消耗。

(2)提高响应速度。

(3)提高线程的可管理性。

Java1.5引入的Executor框架把任务的提交和执行进行解耦,只需要定义好任务,然后提交给线程池,而不用关心该任务是如何执行、被哪个线程执行,以及什么时

候执行。因此,我们可以通过如下方式创建不同形式的线程池。

(1)ExecutorService executor = Executors.newFixedThreadPool(5)

        创建一个指定工作线程数量的线程池。每当提交一个任务就创建一个工作线程,如果工作线程数量达到线程池初始的最大数,则将提交的任务存入到池队列中。

(2)ExecutorService executor = Executors.newCachedThreadPool()

        创建一个可根据需要创建新线程的线程池,但是在以前构造的线程可用时将重用它们。对于执行很多短期异步任务的程序而言,这些线程池通常可提高程序性能。

(3)ExecutorService executor = Executors.newScheduledThreadPool(5)

        创建一个线程池,它可安排在给定延迟后运行命令或者定期地执行。

(4)ExecutorService executor = Executors.newSingleThreadExecutor()

        创建一个使用单个 worker 线程的 Executor,以无界队列方式来运行该线程。

二、内部实现

下面以newFixedThreadPool为例,来看一下线程池内部是怎么实现的。进入newFixedThreadPool,其内部实现如下:

public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}

我们可以发现其只是new了一个ThreadPoolExecutor执行器而已,因此我们也可以自己new一个ThreadPoolExecutor,来达到自己的参数可控的程度,例如,可以将

LinkedBlockingQueue换成其它的(如:SynchronousQueue),只是可读性会降低。

接下来,我们进入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;
}

我们将分别对这几个参数进行介绍:

(1)corePoolSize:线程池核心线程数量,也就是当超过这个范围的时候,就需要将新的Thread放入到等待队列中了。

(2)maximumPoolSize:线程池最大线程数量。一般你用不到,当大于了这个值就会将Thread由一个丢弃处理机制来处理,但是当你发生:newFixedThreadPool的时候,

        corePoolSize和maximumPoolSize是一样的,而corePoolSize是先执行的,所以他会先被放入等待队列,而不会执行到下面的丢弃处理中,看了后面的代码你就知道了。

(3)keepAliveTime:当活跃线程数大于核心线程数时,空闲的多余线程最大存活时间,cachedPoolSize是默认60s,不推荐使用。

(4)workQueue:等待队列,当达到corePoolSize的时候,就向该等待队列放入线程信息。

(5)threadFactory:是构造Thread的方法,你可以自己去包装和传递,主要实现newThread方法即可。

(6)handler:超出线程范围和队列容量的任务的丢弃处理方法。java提供了5种丢弃处理的方法,当然你也可以自己弄,主要是要实现接口:RejectedExecutionHandler。

 

通过如上方式创建好线程以后,我们就可以同execute或者submit方法往线程池中添加具体的执行任务了。我们看submit方法的代码实现:

public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}

可以发现submit方法最终会调用execute方法来进行操作,只是他提供了一个Future来托管返回值的处理而已,当你调用需要有返回值的信息时,你用它来处理是比较

好的,这个Future会包装对Callable信息。因此,我们主要来看一下execute方法的内部实现是怎么样。

public void execute(Runnable command) {
if (command == null)
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);
}

主要分为三步:

(1)如果运行的线程小于corePoolSize,则尝试使用用户定义的Runnalbe对象创建一个新的线程,然后调用调用addWorker函数会原子性的检查runState和workCount,

         通过返回false来防止在不应该添加线程时添加了线程。

(2)如果一个任务能够成功入队列,在添加一个线城时仍需要进行双重检查(因为在前一次检查后该线程死亡了),或者当进入到此方法时,线程池已经shutdown了,

         所以需要再次检查状态,若有必要,当停止时还需要回滚入队列操作,或者当线程池没有线程时需要创建一个新线程。

(3)如果无法入队列,那么需要增加一个新线程,如果此操作失败,那么就意味着线程池已经shutdown或者已经饱和了,所以拒绝任务。

 

更深入的分析addwork函数的内部实现,以及从阻塞队列中getTask任务等其他函数,可以参考博文:http://blog.csdn.net/qq_22929803/article/details/52347381。谢谢!


 

 

 


推荐阅读