Java ThreadPoolExecutorソースノート
概要速記
インタフェースExcutor->インタフェースExutorServices->抽象クラスAbstractExcutorServices->クラスThreadPoolExcutor
スレッド上限ポリシーcorePollSize->blockQueue->maxPollSize->handle
blockQueueパラメータLinkedBlockQueueを使用するとmaxPollSizeパラメータが無効になりますこれは無境界キューです
一般に境界キューArrayBlockQueueを使用する
コアスレッド外のkeepalive timeを設定するパラメータがあり、タイムアウトスレッドは破棄されます.
ThreadPoolExecutor分析
部分フィールド解析 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); Integer.SIZE-3および-1(-1のバイナリ表示を使用して、コンピュータの世界で符号を補って表示した場合、符号ビットを除いてビット毎に逆にして1:11111111111111111111111111111111111111111111111111111111111111111111111111111を加える.シフトして得られたRUNNINGのバイナリは11100000000000000000000000 ctlOfがreturn RUNNING|0であるためctlの初期値も11100000000000000000 AtomicIntegerはJavaでCASメカニズムを用いて実現された原子操作を保証するIntegerクラスレビュークラス注釈 がある. private static final int CAPACITY = (1 << COUNT_BITS) - 1;CAPACITYは1左シフト29ビット減1(29個1)であり,この定数はworkerCountの上限値,(2^29)−1=536870911を表す. 以下は、バイナリ上位に格納スレッドプール状態の定数コード である. private final HashSet workers = new HashSet(); WorkerクラスはThreadPoolExecutorの内部クラスです.このクラスはRunnableインタフェースを実現しました.スレッドプール内のスレッドのキャリアです.新しいスレッドはWorkerを追加します.同時にWorkerクラスはAbstractQueuedSynchronizerという抽象クラスを継承しています.この抽象クラスは双方向チェーンテーブルに基づいて実現される両端キューに基づいてロックを実現する柔軟な制御です.まず、再ロック可能と再ロック不可を理解する必要があります.具体的には、ここをクリックしてこの抽象クラスを実現できる理由は、スレッドがタスクを実行するときに再ロックできないためです.setCorePoolSize()メソッドと同様に、タスクを実行しているスレッドを中断し、このスレッドがアイドル状態であるかどうかを判断することもできます. ブロックスレッドキューフィールドprivate final BlockingQueue workQueue;
ぶぶんほうほうぶんせき構築方法 を直接放出 public Future submit(Callable task)ThreadPoolExecutorというクラス自体のコードファイルには、このsubmitはありませんが、抽象クラスAbstractExecutorServiceを継承しているため、submitメソッドはAbstractExecutroServiceにデフォルト実装されており、ThreadPoolExecutorは自然に継承されています.この方法は主にcallableインタフェースを実現するための戻り値のあるスレッドが である. public void execute(Runnable command)この方法は、上記のスレッドの新規作成メカニズムを処理し、コアプールよりも小さく、スレッドプールがrunning状態にある場合、addWorkerは、コアプールがいっぱいでqueueに参加できる場合、queueでaddWorkerは、同時実行条件で変化するため、単一の怠け者モードでの二重検査と同様に、ここでも、二重チェック が呼び出される. private boolean addWorker(Runnable firstTask,boolean core)coreパラメータがtrueであることは、スレッドの追加時に現在のアクティブスレッド数がcorePoolSizeより少ないかどうかを判断し、falseは、スレッドの追加前に現在のアクティブスレッド数がmaximumPoolSizeより少ないかどうかを判断する必要があることを示します.また、shutdownNowメソッドなどのスレッドプールのライフサイクルを制御する方法もあります.(すべてのスレッドを即座に中断)、interruptIdleWorkersメソッド、shutdownメソッド(忙しいスレッドの実行を待つ、空きスレッドを中断する)など、具体的にはここでは詳しくは説明しませんが、自分で を見てみましょう.
Ref https://www.jianshu.com/p/d2729853c4da
インタフェースExcutor->インタフェースExutorServices->抽象クラスAbstractExcutorServices->クラスThreadPoolExcutor
スレッド上限ポリシーcorePollSize->blockQueue->maxPollSize->handle
blockQueueパラメータLinkedBlockQueueを使用するとmaxPollSizeパラメータが無効になりますこれは無境界キューです
一般に境界キューArrayBlockQueueを使用する
コアスレッド外のkeepalive timeを設定するパラメータがあり、タイムアウトスレッドは破棄されます.
ThreadPoolExecutor分析
部分フィールド解析
* The main pool control state, ctl, is an atomic integer packing
* two conceptual fields
* workerCount, indicating the effective number of threads
* runState, indicating whether running, shutting down etc
* In order to pack them into one int, we limit workerCount to
* (2^29)-1 (about 500 million) threads rather than (2^31)-1 (2
* billion) otherwise representable.
* RUNNING -> SHUTDOWN
* On invocation of shutdown(), perhaps implicitly in finalize()
* (RUNNING or SHUTDOWN) -> STOP
* On invocation of shutdownNow()
* SHUTDOWN -> TIDYING
* When both queue and pool are empty
* STOP -> TIDYING
* When pool is empty
* TIDYING -> TERMINATED
* When the terminated() hook method has completed
// runState is stored in the high-order bits
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;
private static final int TERMINATED = 3 << COUNT_BITS;
ぶぶんほうほうぶんせき
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue 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;
}
RunnableFuture ftask = newTaskFor(task);
execute(ftask);
return ftask;
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);
}
Ref