Java 多线程：线程池实现原理

时间  2019-12-05
标签 java 多线程 线程 实现 原理 栏目 Java 繁體版
原文   原文链接

前言

文章主要来自：点这里 。这也是博主的博客，主要分享了本身接触过的一些后端技术，有不对的地方但愿能够提出。后端

线程池的相关类

咱们都知道，所谓线程池，那么就是至关于有一个池子，线程就放在这个池子中进行重复利用，可以减去了线程的建立和销毁所带来的代价。可是这样并不能很好的解释线程池的原理，下面从代码的角度分析一下线程池的实现。性能

对于原理，在 Java 中，有几个接口，类 值得咱们关注：this

  • Executor线程

  • ExecutorServicecode

  • AbstractExecutorService对象

  • ThreadPoolExecutor继承

Executor

public interface Executor {
    void execute(Runnable command);
}

Executor 接口只有一个 方法，execute，而且须要 传入一个 Runnable 类型的参数。那么它的做用天然是 具体的执行参数传入的任务。接口

ExecutorService

public interface ExecutorService extends Executor {

    void shutdown();

    List<Runnable> shutdownNow();

    boolean isShutdown();

    <T> Future<T> submit(Callable<T> task);

    <T> Future<T> submit(Runnable task, T result);

    Future<?> submit(Runnable task);
    
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException;
    ......
}

ExecutorService 接口继承了 Executor，而且提供了一些其余的方法，好比说：队列

  • shutdownNow ： 关闭线程池，返回放入了线程池，可是还没开始执行的线程。rem

  • submit ： 执行的任务 容许拥有返回值。

  • invokeAll ： 运行把任务放进集合中，进行批量的执行，而且能有返回值

这三个方法也能够说是这个接口重点扩展的方法。

Ps：execute 和 submit 区别：

  • submit 有返回值，execute 没有返回值。 因此说能够根据任务有无返回值选择对应的方法。

  • submit 方便异常的处理。 若是任务可能会抛出异常，并且但愿外面的调用者可以感知这些异常，那么就须要调用 submit 方法，经过捕获 Future.get 抛出的异常。

AbstractExecutorService

AbstractExecutorService 是一个抽象类，主要完成了 对 submit 方法，invokeAll 方法 的实现。 可是其实它的内部仍是调用了 execute 方法，例如：

public Future<?> submit(Runnable task) {
    if (task == null) throw new NullPointerException();
        RunnableFuture<Void> ftask = newTaskFor(task, null);
        execute(ftask);
        return ftask;
    }

ThreadPoolExecutor

ThreadPoolExecutor 继承了 AbstractExecutorService，而且实现了最重要的 execute 方法，是咱们主要须要研究的类。另外，整个线程池是如何实现的呢？

在该类中，有两个成员变量 很是的重要：

private final HashSet<Worker> workers = new HashSet<Worker>();
private final BlockingQueue<Runnable> workQueue;

对于 workers 变量，主要存在了线程对象 Worker，Worker 实现了 Runnable 接口。而对于 workQueue 变量，主要存放了须要执行的任务。 这样其实能够猜到， 整个线程池的实现原理应该是 workQueue 中不断的取出须要执行的任务，放在 workers 中进行处理。

另外，当线程池中的线程用完了以后，多余的任务会等待，那么这个等待的过程是 怎么实现的呢？ 其实若是熟悉 BlockingQueue，那么就会立刻知道，是利用了 BlockingQueue 的take 方法进行处理。

下面具体代码分析：

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

首先，这里须要先理解两个概念。咱们在建立线程池的时候，一般会指定两个变量，一个是maximumPoolSize，另一个是 corePoolSize。

  • 对于 maximumPoolSize：指的是 线程池中最多容许有多少个线程。

  • 对于 corePoolSize： 指的是线程池中正在运行的线程。

在 线程池中，有这样的设定，咱们加入一个任务进行执行，

  • 若是如今线程池中正在运行的线程数量大于 corePoolSize 指定的值而 小于maximumPoolSize 指定的值，那么就会建立一个线程对该任务进行执行，一旦一个线程被建立运行。

  • 若是线程池中的线程数量大于corePoolSize，那么这个任务执行完毕后，该线程会被回收；若是 小于corePoolSize，那么该线程即便空闲，也不会被回收。下个任务过来，那么就使用这个空闲线程。

对于上述代码，首先有：

if (workerCountOf(c) < corePoolSize)

也就是说，判断如今的线程数量是否小于corePoolSize，若是小于，那么就建立一个线程执行该任务，也就是执行

addWorker(command, true)

若是大于，那么就把该任务放进队列当中，即

workQueue.offer(command)

那么，addWorker 是干什么的呢？

private boolean addWorker(Runnable firstTask, boolean core) {    
    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 {
                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;
                    }
             ......
    }

在这里能够看到一些关键代码，例如 w = new Worker(firstTask)， 以及 workers.add(w); 从这里 咱们就能够看到，建立 线程对象 而且加入到 线程 队列中。可是，咱们如今尚未看到具体是怎么执行任务的，继续追踪
w = new Worker(firstTask)，以下代码：

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    {
        ......
        
        final Thread thread;
        
        Runnable firstTask;
        Worker(Runnable firstTask) {
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }
        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 ((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);
                    }
                } 
                ......
    }

在这段代码中，就有不少关键的信息，好比说，Runnable task = w.firstTask;若是为空，那么就 执行 task = getTask()，若是不为空，那么就 进行 task.run(); 调用其方法，这里也就是具体的执行的任务。

如今知道了是怎么样执行具体的任务，那么假如任务的数量 大于 线程池的数量，那么是怎么实现等待的呢，这里就须要看到getTask() 的具体实现了，以下：

private Runnable getTask() {
        for (;;) {
           ......
            try {
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }

这里能够看到， 一个 for 死循环，以及

Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take();

而 workQueue 是 BlockingQueue 类型，也就是带有阻塞的功能。

这就是 线程如何等待执行的。

总结

如今就能够知道，大体的线程池实现原理：

首先，各自存放线程和任务，其中，任务带有阻塞。

private final HashSet<Worker> workers = new HashSet<Worker>();
private final BlockingQueue<Runnable> workQueue;

而后，在 execute 方法中 进行 addWorker(command，true)，也就是建立一个线程，把任务放进去执行；或者是直接把任务放入到任务队列中。

接着 若是是 addWorker，那么就会 new Worker(task) -》调用其中 run() 方法，在Worker 的run() 方法中，调用 runWorker(this); 方法 -》在该方法中就会具体执行咱们的任务 task.run(); 同时这个 runWorker方法至关因而个死循环，正常状况下就会一直取出 任务队列中的任务来执行，这就保证了线程 不会销毁。

因此，这也是为何常说的线程池能够避免线程的频繁建立和 销毁带来的性能消耗。

写在最后

  1. 写出来，说出来才知道对不对，知道不对才能改正，改正了才能成长。

  2. 在技术方面，但愿你们眼里都容不得沙子。若是有不对的地方或者须要改进的地方但愿能够指出，万分感谢。

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程