背景: コンピュータ ハードウェアのアップグレードにより、当社のソフトウェアはマルチスレッド タスクを実行できるようになりました。マルチスレッド プログラミングを行う場合、スレッドを作成する必要があります。プログラムの同時実行性が非常に高い場合は、多数のスレッドを作成し、各スレッドは短いタスクを実行して終了します。スレッドを頻繁に作成すると、スレッドが発生する可能性があります。スレッドの作成と破棄には追加の消費が必要となるため、システムのパフォーマンスが大幅に低下し、サーバーのオーバーヘッドが増加します。
現時点では、プーリング テクノロジを使用してこの欠陥を最適化できるため、スレッド プールが誕生しました。
プーリング テクノロジの本質は、リソースの再利用を実現し、同時実行性の高いシナリオでのリソースの作成と破棄のオーバーヘッドを削減することです。同時実行数が少ない場合、明らかな利点はありません (リソースは常にシステム メモリを占有します)使用する機会がありません。）。
プーリング テクノロジーの概要: プーリング テクノロジーはいつ登場しますか?プーリング テクノロジは、プログラムの同時実行性が高い場合にプログラムを大幅に最適化し、システム内の接続の頻繁な作成と破棄などの追加のオーバーヘッドを削減できるプログラミング手法です。私たちがよく目にするプーリング テクノロジには、データベース接続プール、スレッド プール、オブジェクト プールなどが含まれます。プーリング技術の特徴は、一部の高コストリソースを特定のプール（メモリ）に保持し、その最小接続数、最大接続数、ブロッキングキュー、オーバーフロールールなどの設定を指定して一元管理を容易にすることです。通常の状況では、監視や強制リサイクルなどのサポート機能も備えています。
プーリング テクノロジーはリソース使用テクノロジーです。一般的な使用シナリオは次のとおりです:
リソースの取得コストが高い場合
リソース要求の頻度が高く、使用されるリソースの総数が少ない場合
パフォーマンスの問題や処理時間の遅延が発生する場合
プーリング テクノロジ リソースの分類:
システムによって呼び出されるシステム リソース (スレッド、プロセス、メモリ割り当てなど)
データベース接続、ソケット接続などのネットワーク通信用のリモート リソース。
スレッド プールは、スレッドの作成とスレッドの破棄による追加のオーバーヘッドを回避するために生まれました。そのため、スレッド プールを定義して作成した後は、自分でスレッドを作成する必要はなく、スレッド プール呼び出しを使用してタスクを実行します。スレッド プールを定義して作成する方法を見てみましょう。
スレッド プールを作成するには、シリーズを提供する Executors を使用できます。スレッドを作成するためのファクトリ メソッドの一部 プールと返されたスレッド プールはすべて、ExecutorService インターフェイスを実装します。
ExecutorService このインターフェイスは Executor インターフェイスのサブクラス インターフェイスであり、より広く使用されており、スレッド プールのライフサイクル管理メソッドを提供し、Future オブジェクトを返します。
つまり、Executor を通じてスレッド プールを作成し、ExecutorService を取得し、ExecutorService
## Executor はいくつかのタイプのスレッド プールを作成できます:
newCachedThreadPool キャッシュ可能なスレッド プールを作成します。過剰な場合は 60 になります。超過したスレッド リソースは数秒後にリサイクルされます。タスク ブックが増加し、スレッドが不足すると、新しいスレッドが作成されます。
newScheduledThreadPool スケジュールされた定期的なタスクの実行をサポートするために、固定長のスレッド プールを作成します。
newSingleThreadExecutor シングルスレッド スレッド プールを作成し、タスクの実行に唯一のスレッドのみを使用します。これにより、タスクは送信された順序で確実に完了します。
ThreadPoolExecutor の 7 つのパラメータ:
(1)corePoolSize コア スレッドの数、コア スレッドは常に保持され、変更されません。破壊されました。
maximumPoolSize スレッドの最大数コア スレッドがタスクのニーズを満たせない場合、システムはタスクを実行するための新しいスレッドを作成します。
keepAliveTime コアスレッド以外のスレッドがアイドル状態である生存時間は破棄されます。
timeUnit スレッドの生存時間の単位を表します。
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 オブジェクトです。
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 サイトの他の関連記事を参照してください。
![[Web フロントエンド] Node.js クイック スタート](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
