Java 线程池分析

2023年12月24日 251次阅读来源: 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线程池
    原文地址: http://www.cnblogs.com/dpains/p/7278642.html
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。