ホームページ > 記事 > Java > Javaスレッドプールの実装原理のソースコード分析

Javaスレッドプールの実装原理のソースコード分析

WBOY転載: 2023-05-09 14:16:071488ブラウズ

スレッドプールの起源

背景: コンピュータハードウェアのアップグレードにより、当社のソフトウェアはマルチスレッドタスクを実行できるようになりました。マルチスレッドプログラミングを行う場合、スレッドを作成する必要があります。プログラムの同時実行性が非常に高い場合は、多数のスレッドを作成し、各スレッドは短いタスクを実行して終了します。スレッドを頻繁に作成すると、スレッドが発生する可能性があります。スレッドの作成と破棄には追加の消費が必要となるため、システムのパフォーマンスが大幅に低下し、サーバーのオーバーヘッドが増加します。

現時点では、プーリングテクノロジを使用してこの欠陥を最適化できるため、スレッドプールが誕生しました。

プーリングテクノロジの本質は、リソースの再利用を実現し、同時実行性の高いシナリオでのリソースの作成と破棄のオーバーヘッドを削減することです。同時実行数が少ない場合、明らかな利点はありません (リソースは常にシステムメモリを占有します)使用する機会がありません。）。

プーリングテクノロジーの概要: プーリングテクノロジーはいつ登場しますか?プーリングテクノロジは、プログラムの同時実行性が高い場合にプログラムを大幅に最適化し、システム内の接続の頻繁な作成と破棄などの追加のオーバーヘッドを削減できるプログラミング手法です。私たちがよく目にするプーリングテクノロジには、データベース接続プール、スレッドプール、オブジェクトプールなどが含まれます。プーリング技術の特徴は、一部の高コストリソースを特定のプール（メモリ）に保持し、その最小接続数、最大接続数、ブロッキングキュー、オーバーフロールールなどの設定を指定して一元管理を容易にすることです。通常の状況では、監視や強制リサイクルなどのサポート機能も備えています。

プーリングテクノロジーはリソース使用テクノロジーです。一般的な使用シナリオは次のとおりです:

リソースの取得コストが高い場合
リソース要求の頻度が高く、使用されるリソースの総数が少ない場合
パフォーマンスの問題や処理時間の遅延が発生する場合

プーリングテクノロジリソースの分類:

システムによって呼び出されるシステムリソース (スレッド、プロセス、メモリ割り当てなど)
データベース接続、ソケット接続などのネットワーク通信用のリモートリソース。

スレッドプールの定義と使用

スレッドプールは、スレッドの作成とスレッドの破棄による追加のオーバーヘッドを回避するために生まれました。そのため、スレッドプールを定義して作成した後は、自分でスレッドを作成する必要はなく、スレッドプール呼び出しを使用してタスクを実行します。スレッドプールを定義して作成する方法を見てみましょう。

オプション 1: Executors (理解のためだけに、オプション 2 を使用することをお勧めします)

スレッドプールを作成するには、シリーズを提供する Executors を使用できます。スレッドを作成するためのファクトリメソッドの一部プールと返されたスレッドプールはすべて、ExecutorService インターフェイスを実装します。

ExecutorService このインターフェイスは Executor インターフェイスのサブクラスインターフェイスであり、より広く使用されており、スレッドプールのライフサイクル管理メソッドを提供し、Future オブジェクトを返します。

つまり、Executor を通じてスレッドプールを作成し、ExecutorService を取得し、ExecutorService

# を通じて非同期タスク (Runnable インターフェイスの実装) を実行します。

## Executor はいくつかのタイプのスレッドプールを作成できます:

newCachedThreadPool キャッシュ可能なスレッドプールを作成します。過剰な場合は 60 になります。超過したスレッドリソースは数秒後にリサイクルされます。タスクブックが増加し、スレッドが不足すると、新しいスレッドが作成されます。
newFixedThreadPool は、同時スレッドの最大数を制御できる固定長のスレッドプールを作成します。超過したスレッドはキューで待機します。
newScheduledThreadPool スケジュールされた定期的なタスクの実行をサポートするために、固定長のスレッドプールを作成します。
newSingleThreadExecutor シングルスレッドスレッドプールを作成し、タスクの実行に唯一のスレッドのみを使用します。これにより、タスクは送信された順序で確実に完了します。

オプション 2: ThreadPoolExecutor

Alibaba の開発仕様では、スレッドプールは Executor を介して作成することはできませんが、ThreadPoolExecutor を介して作成されると規定されています。

利点: 書き込みを行う学生は、スレッドプールの実行ルールをより明確に把握でき、リソース枯渇のリスクを回避できます。

ThreadPoolExecutor の 7 つのパラメータ:

(1)

corePoolSize コアスレッドの数、コアスレッドは常に保持され、変更されません。破壊されました。

(2)

maximumPoolSize スレッドの最大数コアスレッドがタスクのニーズを満たせない場合、システムはタスクを実行するための新しいスレッドを作成します。

(3)

keepAliveTime コアスレッド以外のスレッドがアイドル状態である生存時間は破棄されます。

(4)

timeUnit スレッドの生存時間の単位を表します。

(5)

BlockingQueue ブロッキングキュー

実行中のタスクが最大スレッド数を超えた場合、キューに格納できます。スレッドプールに空きリソースがある場合、タスクはキューから削除され、実行を継続できます。
キュータイプには次のタイプが含まれます: LinkedBlockingQueue ArrayBlockingQueue SynchronousQueue TransferQueue。

(6)threadFactory スレッドファクトリは、スレッドの作成に使用されます。スレッドをカスタマイズできます。たとえば、スレッドグループ名を定義できます。これは、jstack の問題のトラブルシューティングを行うときに非常に役立ちます. .

(7)rejectedExecutionHandler 拒否戦略、

すべてのスレッド (最大スレッド数) がビジーで、タスクキューがタスクでいっぱいの場合、拒否が実行されます。。

JDK は 4 つの拒否戦略を提供します。誰もがそれらに精通している必要があります。

AbortPolicy: タスクの削除, そして例外 RejectedExecutionException をスローします。 Default
DiscardPolicy: 例外をスローせずに最新のタスクを破棄します。
DiscardOldestPolicy: キュー時間が最も長いタスク、つまり最も古いタスクを破棄します。
CallerRuns: タスクは呼び出し元 (非同期タスクを送信したスレッド) によって処理されます。

スレッドプールの実装原理

スレッドプールを実装したい場合は、ThreadPoolExecutor クラスに注意する必要があります。Executor は、新しい ThreadPoolExecutor オブジェクト。

ThreadPoolExecutor のクラス継承関係を見ると、Executors によって作成されたスレッドプールが結果 ExecutorService を返す理由がわかります。これは、ThreadPoolExecutor が、 ExecutorService インターフェイス Executor によって作成されるスレッドプールの本質も、作成される ThreadPoolExecutor オブジェクトです。

Javaスレッドプールの実装原理のソースコード分析

ThreadPoolExecutor のソースコードを見てみましょう。まず、ThreadPoolExecutor で定義されている変数と定数は次のとおりです。 ##

    // 复合类型变量 是一个原子整数  控制状态（运行状态|线程池活跃线程数量）
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); 
    private static final int COUNT_BITS = Integer.SIZE - 3; // 低29位 
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1; // 容量
    // 运行状态存储在高位3位
    private static final int RUNNING    = -1 << COUNT_BITS;  // 接受新任务，并处理队列任务
    private static final int SHUTDOWN   =  0 << COUNT_BITS;  // 不接受新任务，但会处理队列任务
    private static final int STOP       =  1 << COUNT_BITS;  // 不接受新任务，不会处理队列任务，中断正在处理的任务
    private static final int TIDYING    =  2 << COUNT_BITS;  // 所有的任务已结束，活跃线程为0，线程过渡到TIDYING状       态，将会执行terminated()钩子方法
    private static final int TERMINATED =  3 << COUNT_BITS;  // terminated()方法已经完成
    // 设置 ctl 参数方法
    private static int runStateOf(int c)     { return c & ~CAPACITY; }
    private static int workerCountOf(int c)  { return c & CAPACITY; }
    private static int ctlOf(int rs, int wc) { return rs | wc; }
    /**
     * 阻塞队列
     */
    private final BlockingQueue<Runnable> workQueue;
    /**
     * Lock 锁.
     */
    private final ReentrantLock mainLock = new ReentrantLock();
    /**
     * 工人们
     */
    private final HashSet<Worker> workers = new HashSet<Worker>();
    /**
     * 等待条件支持等待终止
     */
    private final Condition termination = mainLock.newCondition();
    /**
     * 最大的池大小.
     */
    private int largestPoolSize;
    /**
     * 完成任务数
     */
    private long completedTaskCount;
    /**
     * 线程工厂
     */
    private volatile ThreadFactory threadFactory;
    /**
     * 拒绝策略
     */
    private volatile RejectedExecutionHandler handler;
    /**
     * 存活时间
     */
    private volatile long keepAliveTime;
    /**
     * 允许核心线程数
     */
    private volatile boolean allowCoreThreadTimeOut;
    /**
     * 核心线程数
     */
    private volatile int corePoolSize;
    /**
     * 最大线程数
     */
    private volatile int maximumPoolSize;
    /**
     * 默认拒绝策略
     */
    private static final RejectedExecutionHandler defaultHandler =
        new AbortPolicy();
    /**
     * shutdown and shutdownNow权限
     */
    private static final RuntimePermission shutdownPerm =
        new RuntimePermission("modifyThread");

コンストラクターは、少なくとも 5 つのパラメーターと最大 7 つのパラメーターを持つ 4 つのコンストラクターをサポートします。

    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             threadFactory, defaultHandler);
    }
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              RejectedExecutionHandler handler) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), handler);
    }
    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;
    }

ワーカー、スレッドプールはこれらのワーカーを通じてタスクを実行し、次のような作業を実行します。コア従業員 (コアスレッド) と臨時従業員 (人材が不足している場合に一時的に作成され、アイドル状態の場合は解雇されます)。

    private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    {
        private static final long serialVersionUID = 6138294804551838833L;
        // 工人的本质就是个线程
        final Thread thread;
        // 第一件工作任务
        Runnable firstTask;
      volatile long completedTasks;
        /**
         * 构造器
         */
        Worker(Runnable firstTask) {
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }
        /** 工作  */
        public void run() {
            runWorker(this);
        }
        protected boolean isHeldExclusively() {
            return getState() != 0;
        }
        protected boolean tryAcquire(int unused) {
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }
        protected boolean tryRelease(int unused) {
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }
        public void lock()        { acquire(1); }
        public boolean tryLock()  { return tryAcquire(1); }
        public void unlock()      { release(1); }
        public boolean isLocked() { return isHeldExclusively(); }
        void interruptIfStarted() {
            Thread t;
            if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                }
            }
        }
    }

コアメソッドを介して、スレッドプールがタスクを実行します。 (これはスレッドプールの動作原理でもあります):

タスクを確認します
現在のスレッドプールのステータスを取得します
稼働従業員の数がコア従業員の数より少ないかどうかを判断
少ない場合は人員を募集し、仕事を整理します
それ以上の場合、待機エリアのタスクがいっぱいかどうかを判断します
そうでない場合、タスクは待機エリアに配置されます
待機エリアが満員の場合は、募集を許可するかどうかを参照してください。募集が許可されている場合は、臨時職員が募集されます。
失敗した場合は、スレッドを参照してください。プールは新しいタスクを受け取ることができず、上司によって合意された拒否ポリシーに従って拒否ポリシーの実行を開始します

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

submit()メソッドは、そのメソッドによって定義されます。抽象親クラスです。ここでは、submit とexecute の違いがはっきりとわかります。submit を呼び出すと、RunnableFuture が作成され、Future が返されます。ここで、submit メソッドに戻り値の型を伝えることができ、それが一般的な制約を通じて値を返します。

public abstract class AbstractExecutorService implements ExecutorService {
    public Future<?> submit(Runnable task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<Void> ftask = newTaskFor(task, null);
        execute(ftask);
        return ftask;
    }
    public <T> Future<T> submit(Runnable task, T result) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task, result);
        execute(ftask);
        return ftask;
    }
    public <T> Future<T> submit(Callable<T> task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task);
        execute(ftask);
        return ftask;
    }
    ...
}

addWorker() は人を募集するメソッドです:

    private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
            // 判断状态，及任务列表
            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
            }
        }
        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;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

タスクを取得するメソッド:

    private Runnable getTask() {
        boolean timedOut = false; // Did the last poll() time out?
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
            // Check if queue empty only if necessary.
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }
            int wc = workerCountOf(c);
            // Are workers subject to culling?
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }
            try {
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }

従業員に仕事をさせ、タスクを割り当て、タスクを実行する方法:

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

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

声明：

この記事はyisu.comで複製されています。侵害がある場合は、admin@php.cn までご連絡ください。

前の記事：Java 文字エンコーディングの例の分析次の記事：Java 文字エンコーディングの例の分析

続きを見る

Javaスレッドプールの実装原理のソースコード分析

スレッド プールの起源

スレッド プールの定義と使用

オプション 1: Executors (理解のためだけに、オプション 2 を使用することをお勧めします)

スレッド プールの実装原理

関連記事

スレッドプールの起源

スレッドプールの定義と使用

スレッドプールの実装原理