首页 > 代码库 > Java 线程池分析

Java 线程池分析

在项目中经常会用到java线程池,但是别人问起线程池的原理,线程池的策略怎么实现的? 答得不太好,所以按照源码分析一番,首先看下最常用的线程池代码:

public class ThreadPoolTest {

    private static Executor executor= Executors.newFixedThreadPool(10);  //一般通过fixThreadPool起线程池

    public static void main(String[] args){

        for(int i=0;i<50;i++){

            executor.execute(new Task());  //通过线程池执行task
        }
    }

    static class Task implements Runnable{  //Task任务,打印当前执行该任务的线程名称


        @Override
        public void run() {

            System.out.println(Thread.currentThread().getName());
        }
    }
}

这个输出结果为,可以看到,没有线程池的超过10的,所以线程池执行只用了10个线程:

pool-1-thread-1
pool-1-thread-2
pool-1-thread-3
pool-1-thread-4
pool-1-thread-5
pool-1-thread-6
pool-1-thread-7
pool-1-thread-8
pool-1-thread-9
pool-1-thread-10
pool-1-thread-10
pool-1-thread-2
pool-1-thread-3
pool-1-thread-5
pool-1-thread-2
pool-1-thread-4
pool-1-thread-2
pool-1-thread-3
pool-1-thread-2

 

接下来分析一下源码,在newFixedThreadPool(10)的时候,源码是这样的,返回一个ThreadPoolExecutor

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

再看下ThreadPoolExecutor的构造方法,如下,参数说明我标在注释中了,默认参数见上面传的。所以在newFixedThreadPool(10)时,其实只传了一个参数,其他的都为默认。

public ThreadPoolExecutor(int corePoolSize,  //线程池核心池大小,当提交一个任务时,线程池会新创建一个线程池来执行任务,直到线程池大小等于该值。等于后继续提交的任务被保存在阻塞队列,等待执行。
                              int maximumPoolSize, //最大线程池大小,当阻塞队列满后,如果继续提交任务,则创建新的线程执行任务,前提是当前线程数小于maximumPoolSize;
                              long keepAliveTime, //线程池keepalive时间,即空闲线程池的存活时间。
                              TimeUnit unit,  //keepAliveTime的单位
                              BlockingQueue<Runnable> workQueue) //阻塞队列,比如实现Runnable接口{
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }

说明:
在JDK中提供了如下阻塞队列:
1、ArrayBlockingQueue:基于数组结构的有界阻塞队列,按FIFO排序任务;
2、LinkedBlockingQuene:基于链表结构的阻塞队列,按FIFO排序任务,吞吐量通常要高于ArrayBlockingQuene;
3、SynchronousQuene:一个不存储元素的阻塞队列,每个插入操作必须等到另一个线程调用移除操作,否则插入操作一直处于阻塞状态,吞吐量通常要高于LinkedBlockingQuene;
4、priorityBlockingQuene:具有优先级的无界阻塞队列;
 
接下来执行this构造方法,查看this构造方法源码,多了两个参数,一个为ThreadFactory(线程池工厂方法),另一个为RejectedExecutionHandler(饱和策略),上面传入的参数为Executors.defaultThreadFactory()和defaultHandler,可以看到defaultThreadFactory方法如下,defaultHandler的策略为abort。
    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() {  //设置线程名,group等
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() :
                                  Thread.currentThread().getThreadGroup();
            namePrefix = "pool-" +
                          poolNumber.getAndIncrement() +
                         "-thread-";
        }

 

private static final RejectedExecutionHandler defaultHandler =  //默认饱和策略
        new AbortPolicy();

 

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

 

了解源码后,接下来要验证一个事情,即提交任务后线程池的执行策略及饱和策略
我们将最开始部分的代码的线程池,自己通过ThreadPoolExecutor的实现,只修改了线程池的实例部门,将maximumPoolSize设置为20,同时将阻塞队列的大小改为长度为3的ArrayBlockingQueue,饱和策略默认不传。这时候还是去提交50个任务。理论上,我们应该会创建20个线程。
public class ThreadPoolTest {
    private static Executor executor1= new ThreadPoolExecutor(10,20,0, TimeUnit.MILLISECONDS,new ArrayBlockingQueue<Runnable>(3));

    public static void main(String[] args){

        for(int i=0;i<50;i++){

            executor.execute(new Task());
        }
    }

    static class Task implements Runnable{


        @Override
        public void run() {

            System.out.println(Thread.currentThread().getName());
        }
    }
}

执行结果如下,确实看到了20个线程的名称:

pool-1-thread-1
pool-1-thread-3
pool-1-thread-2
pool-1-thread-5
pool-1-thread-4
pool-1-thread-6
pool-1-thread-7
pool-1-thread-8
pool-1-thread-9
pool-1-thread-10
pool-1-thread-1
pool-1-thread-1
pool-1-thread-1
pool-1-thread-1
pool-1-thread-11
pool-1-thread-12
pool-1-thread-12
pool-1-thread-12
pool-1-thread-12
pool-1-thread-12
pool-1-thread-13
pool-1-thread-14
pool-1-thread-14
pool-1-thread-14
pool-1-thread-14
pool-1-thread-15
pool-1-thread-16
pool-1-thread-16
pool-1-thread-16
pool-1-thread-16
pool-1-thread-17
pool-1-thread-17
pool-1-thread-17
pool-1-thread-17
pool-1-thread-18
pool-1-thread-18
pool-1-thread-18
pool-1-thread-18
pool-1-thread-19
pool-1-thread-19
pool-1-thread-19
pool-1-thread-19
pool-1-thread-20
pool-1-thread-20
pool-1-thread-20
pool-1-thread-20

 
再接下来我们将maximumPoolSize也改为10,再执行代码,这个时候应该会出现任务提交失败的情况;因为阻塞队列也满了,线程池最多只能创建10个,提交的任务有50个,处理不过来。
结果如下,可以看到完成的task为13个,即10个线程池大小+长度为3的有界阻塞队列,最后使用的策略为abort策略:

pool-1-thread-1
pool-1-thread-3
pool-1-thread-2
pool-1-thread-4
pool-1-thread-5
pool-1-thread-6
pool-1-thread-7
pool-1-thread-8
pool-1-thread-9
pool-1-thread-9
pool-1-thread-9
pool-1-thread-9
pool-1-thread-10
Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task ThreadPoolTest$Task@4dc63996 rejected from java.util.concurrent.ThreadPoolExecutor@d716361[Running, pool size = 10, active threads = 0, queued tasks = 0, completed tasks = 13]
at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047)

 

 

 

线程池的源码及调度流程分析完成了,接下来看下线程池的executor怎么去执行的
先看下线程的状态
其中AtomicInteger变量ctl的功能非常强大:利用低29位表示线程池中线程数,通过高3位表示线程池的运行状态:
1、RUNNING:-1 << COUNT_BITS,即高3位为111,该状态的线程池会接收新任务,并处理阻塞队列中的任务;
2、SHUTDOWN: 0 << COUNT_BITS,即高3位为000,该状态的线程池不会接收新任务,但会处理阻塞队列中的任务;
3、STOP : 1 << COUNT_BITS,即高3位为001,该状态的线程不会接收新任务,也不会处理阻塞队列中的任务,而且会中断正在运行的任务;
4、TIDYING : 2 << COUNT_BITS,即高3位为010;
5、TERMINATED: 3 << COUNT_BITS,即高3位为011;
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();  //获取ctl的低29位
        if (workerCountOf(c) < corePoolSize) { //workerCountOf方法根据ctl的低29位,得到线程池的当前线程数,如果当前的线程数小于核心池大小,则直接addWorker
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        if (isRunning(c) && workQueue.offer(command)) { //如果当前线程池处于Running状态(即能够接受任务,处理阻塞队列中的任务),且把任务提交给了阻塞队列
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))  //再次检查线程池的状态,如果线程池没有RUNNING(即无法放入队列),并且将任务从队列移除,交给reject处理
                reject(command);
            else if (workerCountOf(recheck) == 0)  //否则,如果线程池数为0,则添加worker,启动一个null任务的线程
                addWorker(null, false);
        }
        else if (!addWorker(command, false))   //否则如果添加addWork失败,则reject。
            reject(command);
    }

 

addWorker主要负责创建新的线程并执行任务,主代码如下:
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 startable
                            throw 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;

 

线程池的工作线程通过Woker类实现,在ReentrantLock锁的保证下,把Woker实例插入到HashSet后,并启动Woker中的线程,其中Worker类设计如下:
1、继承了AQS类,可以方便的实现工作线程的中止操作;
2、实现了Runnable接口,可以将自身作为一个任务在工作线程中执行;
3、当前提交的任务firstTask作为参数传入Worker的构造方法;

private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable

Worker(Runnable firstTask) { setState(
-1); // inhibit interrupts until runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } /** Delegates main run loop to outer runWorker */ public void run() { runWorker(this); }

线程工厂在创建线程thread时,将Woker实例本身this作为参数传入,当执行start方法启动线程thread时,本质是执行了Worker的runWorker方法。

 

 

在线程池执行任务的时候,除了executor还有一种方法叫做submit,通过ExecutorService.submit()方法提交的任务,可以获取任务执行完的返回值。

在实际业务场景中,Future和Callable基本是成对出现的,Callable负责产生结果,Future负责获取结果。
1、Callable接口类似于Runnable,只是Runnable没有返回值。
2、Callable任务除了返回正常结果之外,如果发生异常,该异常也会被返回,即Future可以拿到异步执行任务各种结果;
3、Future.get方法会导致主线程阻塞,直到Callable任务执行完成;

 








Java 线程池分析