Java スレッド プールの実装原理を 1 つの記事で理解する

この記事では、java に関する関連知識を提供します。主に、スレッド プールを使用する理由やスレッド プールの使用に関する関連コンテンツなど、スレッド プールの実装原理に関する関連コンテンツを紹介します。それを見て、皆さんのお役に立てれば幸いです。

推奨学習: 「

java ビデオ チュートリアル 」

1. スレッド プールを使用する理由

スレッド プールの使用は通常、次の 2 つの理由によるものです。

1. 頻繁にスレッドを作成および破棄するとシステム リソースが消費され、スレッド プールを使用するとスレッドを再利用できます。

2. スレッド プールを使用すると、スレッドの数を動的に管理したり、ブロッキング キューを設定したり、タスクを定期的に実行したり、環境を分離したりすることができ、スレッドの管理が容易になります。

#2. スレッド プールの使用法

/**
 * @author 一灯架构
 * @apiNote 线程池示例
 **/
public class ThreadPoolDemo {

    public static void main(String[] args) {
        // 1. 创建线程池
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                3,
                3,
                0L,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue(),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());
      
        // 2. 往线程池中提交3个任务
        for (int i = 0; i  {
                System.out.println(Thread.currentThread().getName() + " 关注公众号：一灯架构");
            });
        }
      
        // 3. 关闭线程池
        threadPoolExecutor.shutdown();
    }
}

出力結果:

pool-1-thread-2 关注公众号：一灯架构
pool-1-thread-1 关注公众号：一灯架构
pool-1-thread-3 关注公众号：一灯架构

スレッド プールの使用法は非常に簡単です:

新しい ThreadPoolExecutor() コンストラクターを呼び出し、コア パラメーターを指定して、スレッド プールを作成します。
execute() メソッドを呼び出して実行可能タスクを送信します。
使用後は、shutdown() メソッドを呼び出してスレッド プールを閉じます。
スレッド プール構築方法におけるコア パラメーターの役割をもう一度見てみましょう。

3. スレッド プールのコア パラメータ

スレッド プールには 7 つのコア パラメータがあります:

パラメータ名int corePoolSizeint minimumPoolSize long keepAliveTimeTimeUnit単位##ブロッキングキュー#ThreadFactory threadFactoryスレッド作成ファクトリRejectedExecutionHandler ハンドラー拒否戦略

• corePoolSize 核心线程数

  当往线程池中提交任务，会创建线程去处理任务，直到线程数达到corePoolSize，才会往阻塞队列中添加任务。默认情况下，空闲的核心线程并不会被回收，除非配置了allowCoreThreadTimeOut=true。

• maximumPoolSize 最大线程数

  当线程池中的线程数达到corePoolSize，阻塞队列又满了之后，才会继续创建线程，直到达到maximumPoolSize，另外空闲的非核心线程会被回收。

• keepAliveTime 线程存活时间

  非核心线程的空闲时间达到了keepAliveTime，将会被回收。

• TimeUnit 时间单位

  线程存活时间的单位，默认是TimeUnit.MILLISECONDS（毫秒），可选择的有：

  TimeUnit.NANOSECONDS（纳秒） TimeUnit.MICROSECONDS（微秒） TimeUnit.MILLISECONDS（毫秒） TimeUnit.SECONDS（秒） TimeUnit.MINUTES（分钟） TimeUnit.HOURS（小时） TimeUnit.DAYS（天）

• workQueue 阻塞队列

  当线程池中的线程数达到corePoolSize，再提交的任务就会放到阻塞队列的等待，默认使用的是LinkedBlockingQueue，可选择的有：

  LinkedBlockingQueue（基于链表实现的阻塞队列）

  ArrayBlockingQueue（基于数组实现的阻塞队列）

  SynchronousQueue（只有一个元素的阻塞队列）

  PriorityBlockingQueue（实现了优先级的阻塞队列）

  DelayQueue（实现了延迟功能的阻塞队列）

• threadFactory 线程创建工厂

  用来创建线程的工厂，默认的是Executors.defaultThreadFactory()，可选择的还有Executors.privilegedThreadFactory()实现了线程优先级。当然也可以自定义线程创建工厂，创建线程的时候最好指定线程名称，便于排查问题。

• RejectedExecutionHandler 拒绝策略

  当线程池中的线程数达到maximumPoolSize，阻塞队列也满了之后，再往线程池中提交任务，就会触发执行拒绝策略，默认的是AbortPolicy（直接终止，抛出异常），可选择的有：

  AbortPolicy（直接终止，抛出异常）

  DiscardPolicy（默默丢弃，不抛出异常）

  DiscardOldestPolicy（丢弃队列中最旧的任务，执行当前任务）

  CallerRunsPolicy（返回给调用者执行）

4. 线程池工作原理

线程池的工作原理，简单理解如下：

• 当往线程池中提交任务的时候，会先判断线程池中线程数是否核心线程数，如果小于，会创建核心线程并执行任务。

• 如果线程数大于核心线程数，会判断阻塞队列是否已满，如果没有满，会把任务添加到阻塞队列中等待调度执行。

• 如果阻塞队列已满，会判断线程数是否小于最大线程数，如果小于，会继续创建最大线程数并执行任务。

• 如果线程数大于最大线程数，会执行拒绝策略，然后结束。

5. 线程池源码剖析

5.1 线程池的属性

public class ThreadPoolExecutor extends AbstractExecutorService {

    // 线程池的控制状态，Integer长度是32位，前3位用来存储线程池状态，后29位用来存储线程数量
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    // 线程个数所占的位数
    private static final int COUNT_BITS = Integer.SIZE - 3;
    // 线程池的最大容量，2^29-1，约5亿个线程
    private static final int CAPACITY = (1  workers = new HashSet();
    // 等待条件，用来响应中断
    private final Condition termination = mainLock.newCondition();
    // 是否允许回收核心线程
    private volatile boolean allowCoreThreadTimeOut;
    // 线程数的历史峰值
    private int largestPoolSize;

    /**
     * 以下是线程池的七大核心参数
     */
    private volatile int corePoolSize;
    private volatile int maximumPoolSize;
    private volatile long keepAliveTime;
    private final BlockingQueue<runnable> workQueue;
    private volatile ThreadFactory threadFactory;
    private volatile RejectedExecutionHandler handler;

}</runnable>

线程池的控制状态ctl用来存储线程池状态和线程个数，前3位用来存储线程池状态，后29位用来存储线程数量。

设计者多聪明，用一个变量存储了两块内容。

5.2 线程池状态

线程池共有5种状态：

パラメータの意味
コア スレッドの数
スレッドの最大数
スレッド生存時間
時間単位	#BlockingQueue workQueue

状态名称	状态含义	状态作用
RUNNING	运行中	线程池创建后默认状态，接收新任务，并处理阻塞队列中的任务。
SHUTDOWN	已关闭	调用shutdown方法后处于该状态，不再接收新任务，处理阻塞队列中任务。
STOP	已停止	调用shutdownNow方法后处于该状态，不再新任务，并中断所有线程，丢弃阻塞队列中所有任务。
TIDYING	处理中	所有任务已完成，所有工作线程都已回收，等待调用terminated方法。
TERMINATED	已终止	调用terminated方法后处于该状态，线程池的最终状态。

5.3 execute源码

看一下往线程池中提交任务的源码，这是线程池的核心逻辑：

// 往线程池中提交任务
public void execute(Runnable command) {
    // 1. 判断提交的任务是否为null
    if (command == null)
        throw new NullPointerException();

    int c = ctl.get();
    // 2. 判断线程数是否小于核心线程数
    if (workerCountOf(c) <p>execute方法的逻辑也很简单，最终就是调用addWorker方法，把任务添加到worker集合中，再看一下addWorker方法的源码：</p><pre class="brush:php;toolbar:false">// 添加worker
private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (; ; ) {
        int c = ctl.get();
        int rs = runStateOf(c);
        // 1. 检查是否允许提交任务
        if (rs >= SHUTDOWN &&
                !(rs == SHUTDOWN &&
                        firstTask == null &&
                        !workQueue.isEmpty()))
            return false;
        // 2. 使用死循环保证添加线程成功
        for (; ; ) {
            int wc = workerCountOf(c);
            // 3. 校验线程数是否超过容量限制
            if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // 4. 使用CAS修改线程数
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();
            // 5. 如果线程池状态变了，则从头再来
            if (runStateOf(c) != rs)
                continue retry;
        }
    }
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        // 6. 把任务和新线程包装成一个worker
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            // 7. 加锁，控制并发
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // 8. 再次校验线程池状态是否异常
                int rs = runStateOf(ctl.get());
                if (rs  largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                // 12. 启动线程
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (!workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

方法虽然很长，但是逻辑很清晰。就是把任务和线程包装成worker，添加到worker集合，并启动线程。

5.4 worker源码

再看一下worker类的结构：

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable {
    // 工作线程
    final Thread thread;
    // 任务
    Runnable firstTask;

    // 创建worker，并创建一个新线程（用来执行任务）
    Worker(Runnable firstTask) {
        setState(-1);
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }
}

5.5 runWorker源码

再看一下run方法的源码：

// 线程执行入口
public void run() {
    runWorker(this);
}

// 线程运行核心方法
final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock();
    boolean completedAbruptly = true;
    try {
        // 1. 如果当前worker中任务是null，就从阻塞队列中获取任务
        while (task != null || (task = getTask()) != null) {
            // 加锁，保证thread不被其他线程中断（除非线程池被中断）
            w.lock();
            // 2. 校验线程池状态，是否需要中断当前线程
            if ((runStateAtLeast(ctl.get(), STOP) ||
                    (Thread.interrupted() &&
                            runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                wt.interrupt();
            try {
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    // 3. 执行run方法
                    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 {
        // 4. 从worker集合删除当前worker
        processWorkerExit(w, completedAbruptly);
    }
}

runWorker方法逻辑也很简单，就是不断从阻塞队列中拉取任务并执行。

再看一下从阻塞队列中拉取任务的逻辑：

// 从阻塞队列中拉取任务
private Runnable getTask() {
    boolean timedOut = false;
    for (; ; ) {
        int c = ctl.get();
        int rs = runStateOf(c);
        // 1. 如果线程池已经停了，或者阻塞队列是空，就回收当前线程
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }
        int wc = workerCountOf(c);
        // 2. 再次判断是否需要回收线程
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }
        try {
            // 3. 从阻塞队列中拉取任务
            Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

推荐学习：《java视频教程》

以上がJava スレッド プールの実装原理を 1 つの記事で理解するの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。