<返回更多

从根儿上认识线程池

2020-05-18    
加入收藏

前言

大家好,我是jack xu,本篇是并发编程的第二篇,今天跟大家聊一聊线程池的那点事。本篇文章有点长,小伙们静下心、耐下心来把他看完。。

为什么要使用线程池

1)降低创建线程和销毁线程的性能开销

2)提高响应速度,当有新任务需要执行是不需要等待线程创建就可以立马执行

3)合理的设置线程池大小可以避免因为线程数超过硬件资源瓶颈带来的问题

 

从根儿上认识线程池

 

我们来看阿里巴巴的代码规范,在项目中创建线程必须要使用线程池创建,原因就是我说的以上三点

 

线程池的使用

首先我们来看下UML类图

 

从根儿上认识线程池

 

 

搞清楚上面的结构很重要,Executors是一个工具类,然后看创建线程的两种方式,第一种是通过Executors提供的工厂方法来实现,有下面四种方式

        Executor executor1 = Executors.newFixedThreadPool(10);
        Executor executor2 = Executors.newSingleThreadExecutor();
        Executor executor3 = Executors.newCachedThreadPool();
        Executor executor4 = Executors.newScheduledThreadPool(10);
复制代码

第二种是通过构造方法来实现

        ExecutorService executor5 = new ThreadPoolExecutor(1,
                1,
                0L,
                TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(2), Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());
复制代码

其实查看第一种方式创建的源码就会发现:

    public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue());
    }
复制代码

根本上还是通过调用ThreadPoolExecutor的构造方法,创建时传入不同参数,所以本质上还是只有一种创建线程池的方式,就是用构造方法,这里我不想讲用Executors的工厂方法具体帮我们创建了怎样的线程池,让我们再来看一条阿里巴巴规范。

 

从根儿上认识线程池

 

看到这里大家都明白了吧,正是因为封装性太强了,反而小伙们会不知道怎么用,乱用,滥用,有可能会导致OOM,除非你对创建的这四个线程池了如指掌,所以我介绍了也是白介绍,因为就不在用,接下来我们重点看下ThreadPoolExecutor构造方法里各个参数的含义,构造方法有很多个,我选了一个最完整的。

 

public ThreadPoolExecutor(int corePoolSize, //核心线程数量
                          int maximumPoolSize, //最大线程数
                          long keepAliveTime, //超时时间,超出核心线程数量以外的线程空余存活时间
                          TimeUnit unit, //存活时间单位
                          BlockingQueue<Runnable> workQueue, //保存执行任务的队列
                          ThreadFactory threadFactory,//创建新线程使用的工厂
                          RejectedExecutionHandler handler //当任务无法执行的时候的处理方式)
复制代码
1)ArrayBlockingQueue:基于数组的先进先出队列,此队列创建时必须指定大小;  
2)LinkedBlockingQueue:基于链表的先进先出队列,如果创建时没有指定此队列大小,则默认为Integer.MAX_VALUE;
3)SynchronousQueue:这个队列比较特殊,它不会保存提交的任务,而是将直接新建一个线程来执行新来的任务。
复制代码
1、AbortPolicy:直接抛出异常,默认策略;
2、CallerRunsPolicy:用调用者所在的线程来执行任务;
3、DiscardOldestPolicy:丢弃阻塞队列中靠最前的任务,并执行当前任务;
4、DiscardPolicy:直接丢弃任务;
当然也可以根据应用场景实现 RejectedExecutionHandler 接口,自定义饱和策略,如记录
日志或持久化存储不能处理的任务
复制代码

创建完线程池后使用也很简单,带返回值和不带返回值,传入对应传入Runnable或者Callable接口的实现

        //无返回值
        executor5.execute(() -> System.out.println("jack xushuaige"));
        //带返回值
        String message = executor5.submit(() -> { return "jack xushuaige"; }).get();
复制代码

源码分析

execute方法

基于源码入口进行分析,先看execute方法

    public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        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,那么尝试创建新的线程,并把传入的 command 作为它的第一个 task 来执行。调用 addWorker 会自动检查 runState 和 workCount,以此来防止在不应该添加线程时添加线程的错误警告;

2、即使 task 可以被成功加入队列,我们仍旧需要再次确认我们是否应该添加 thread(因为最后一次检查之后可能有线程已经死掉了)还是线程池在进入此方法后已经停掉了。所以我们会再次检查状态,如果有必要的话,可以回滚队列。或者当没有线程时,开启新的 thread;

3、如果无法将 task 加入 queue,那么可以尝试添加新的 thread。如果添加失败,这是因为线程池被关闭或者已经饱和了,所以拒绝这个 task。

如果你看完以后还是一脸懵逼,那没事,我把这个流程图画下来,你品,你细品,好好理解一下

从根儿上认识线程池

 

 

然后介绍一下源码中ctl是干什么的,点进去查看源码

 

从根儿上认识线程池

 

我们发现它是一个原子类,主要作用是用来保存线程数量和线程池的状态,他用到了位运算, 一个int数值是32个 bit 位,这里采用高 3 位来保存运行状态,低 29 位来保存线程数量。

 

我们来计算一下ctlOf(RUNNING, 0)方法,其中 RUNNING =-1 << COUNT_BITS ; -1 左移 29 位,-1 的二进制是32个1(1111 1111 1111 1111 1111 1111 1111 1111),左移29位后得到(1110 0000 0000 0000 0000 0000 0000 0000),然后111| 0还是111,同理可得其他状态的 bit 位。这个位运算很有意思,hashmap源码中也用到了位运算,小伙们在平时开发中也可以尝试用下,这样运算速度会快,而且能够装b,介绍下这五种线程池的状态

他们的转换关系如下:

 

从根儿上认识线程池

 

 

addWorker方法

我们看到execute流程的核心方法为addWorker,我们继续分析,源码看起来比较唬人,其实就做了两件事,拆分一下

第一步:更新worker的数量,代码如下:

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 ctl
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
        }
复制代码

retry是一个标记,和循环配合使用,continue retry 的时候,会跳到 retry 的地方再次执行。如果 break retry,则跳出整个循环体。源码先获取到 ctl,然后检查状态,然后根据创建线程类型的不同,进行数量的校验。在通过CAS方式更新 ctl状态,成功的话则跳出循环。否则再次取得线程池状态,如果和最初已经不一致,那么从头开始执行。如果状态并未改变则继续更新worker的数量。流程图如下:

 

从根儿上认识线程池

 

第二步:添加 worker 到 workers 的 set 中。并且启动 worker 中持有的线程。代码如下:

 

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;
复制代码

可以看到添加 work 时需要先获得锁,这样确保多线程并发安全。如果添加 worker 成功,那么调用 worker 中线程的 start 方法启动线程。如果启动失败则调用 addWorkerFailed 方法进行回滚。看到这里小伙们会发现

1、ThreadPoolExecutor在初始化后并没有启动和创建任何线程,在调用 execute方法时才会调用 addWorker创建线程

2、addWorker方法中会创建新的worker,并启动其持有的线程来执行任务。

上文提到如果线程数量已经达到corePoolSize,则只会把command 加入到 workQueue中,那么加入到 workQueue中的command是如何被执行的呢?下面我们来分析 Worker 的源代码。

Worker类

Worker封装了线程,是executor中的工作单元。worker继承自AbstractQueuedSynchronizer,并实现 Runnable。 Worker 简单理解其实就是一个线程,里面重新了 run 方法,我们来看他的构造方法:

        Worker(Runnable firstTask) {
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }
复制代码

再来看下这两个重要的属性

        /** Thread this worker is running in.  Null if factory fails. */
        final Thread thread;
        /** Initial task to run.  Possibly null. */
        Runnable firstTask;
复制代码

firstTask 用它来保存传入的任务;thread 是在调用构造方法时通过 ThreadFactory 来创建的线程,是用来处理任务的线程,这里用的是 ThreadFactory 创建线程,并没有直接 new,原因上文也提到过,这里看下 newThread 传入的是 this,因为 Worker 本身继承了 Runnable 接口,所以 addWork 中调用的 t.start(),实际上运行的是 t 所属 worker 的 run 方法。worker 的 run 方法如下:

public void run() {
    runWorker(this);
}
复制代码

runWorker源码再如下:

    final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
            while (task != null || (task = getTask()) != null) {
                w.lock();
                // If pool is stopping, ensure thread is interrupted;
                // if not, ensure thread is not interrupted.  This
                // requires a recheck in second case to deal with
                // shutdownNow race while clearing interrupt
                if ((runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                    wt.interrupt();
                try {
                    beforeExecute(wt, task);
                    Throwable thrown = null;
                    try {
                        task.run();
                    } catch (RuntimeException x) {
                        thrown = x; throw x;
                    } catch (Error x) {
                        thrown = x; throw x;
                    } catch (Throwable x) {
                        thrown = x; throw new Error(x);
                    } finally {
                        afterExecute(task, thrown);
                    }
                } finally {
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }
复制代码

简单分析一下

1、先取出 worker 中的 firstTask,并清空;

2、如果没有 firstTask,则调用 getTask 方法,从 workQueue 中获取task;

3、获取锁;

4、执行 beforeExecute。这里是空方法,如有需要在子类实现;

5、执行 task.run;

6、执行 afterExecute。这里是空方法,如有需要在子类实现;

7、清空 task,completedTasks++,释放锁;

8、当有异常或者没有 task 可执行时,会进入外层 finnaly 代码块。调用 processWorkerExit 退出当前 worker。从 works 中移除本 worker 后,如果 worker 数量小于 corePoolSize,则