Javaスレッドプールの使用
- 大家讲道理オリジナル
- 2017-05-28 11:33:171586ブラウズ
1. タスクと実行戦略間の暗黙的な結合
エグゼキューターはタスク送信とタスク実行戦略を切り離すことができます
同じタイプで実行時間にほとんど差がないタスクのみが最大のパフォーマンスを達成できます。 - スレッド プール内の消費タスクと消費時間が短いタスク。スレッド プールが非常に大きくない限り、デッドロックやその他の問題が発生します
1. スレッド スターベーション デッドロック
に似ています: 2 つのタスクを送信する 単一スレッド プールがある場合、 2 つのタスクは相互に依存しており、1 つのタスクが別のタスクを待機すると、デッドロックが発生します。プールでは不十分です
定義: タスクはプール内の他のタスクの実行結果を待機する必要があり、スターベーションが発生します。デッドロックが発生する可能性があります
2. スレッドプールのサイズ
注: スレッドプールのサイズは、他のリソースプールなどの他の制限にも影響されます: データベース接続プール
各タスクが接続の場合、 thread プールのサイズはデータベース接続プールのサイズに依存します
3. ThreadPoolExecutor スレッド プールを設定します
Instance:
1. Executors ファクトリ メソッドを通じていくつかのデフォルト実装を返します
2. インスタンス化することによってThreadPoolExecutor (.. ...)
スレッド プールの キューのカスタム実装
1. 無制限のキュー: タスクが到着し、スレッド プールがいっぱいになると、タスクはキュー内で待機します。 , キューは無限に拡張されます
例: これはシングルトンと固定サイズのスレッド プールが使用するものです
2. 境界付きキュー: 新しいタスクが到着してキューがいっぱいの場合は、 飽和戦略を使用します
3. 同期ハンドオーバー: スレッドプールが大きい場合、タスクをキューに入れた後、タスクプロデューサーがすぐにタスクをキューに入れると、ハンドオーバーに遅れが生じます
SynchronousQueue は直接ハンドオーバーします。タスクをワーカースレッドに
メカニズム: タスクを入れます、受け入れを待っているスレッドがあるはずです、そうでない場合はaddthread、スレッドが飽和している場合はタスクを拒否します
例: キャッシュ ThreadPool は使用される戦略です
飽和戦略:
setRejectedExecutionHandlerを使用して飽和戦略を変更します
1. 終了Abort (デフォルト): 例外をスローします 発信者によって処理される
2. 破棄 Discard
3. 破棄 DiscardOldest: 最も古いタスクを破棄します。注: priorityの場合、キューは最も優先度の高いタスクを破棄します
4. CallerRuns:タスクをロールバックすると、呼び出し元のスレッドがそれを自動的に処理します
4. スレッド ファクトリ ThreadFactoy
スレッドが作成されるたびに、実際にスレッド ファクトリを呼び出して完了します
カスタム スレッド ファクトリ: ThreadFactory を実装します
をカスタマイズできますスレッド ファクトリの動作: Uncaught Exception Handler など
public class MyAppThread extends Thread { public static final String DEFAULT_NAME = "MyAppThread"; private static volatile boolean debugLifecycle = false; private static final AtomicInteger created = new AtomicInteger(); private static final AtomicInteger alive = new AtomicInteger(); private static final Logger log = Logger.getAnonymousLogger(); public MyAppThread(Runnable r) { this(r, DEFAULT_NAME);
} public MyAppThread(Runnable runnable, String name) { super(runnable, name + "-" + created.incrementAndGet()); //设置该线程工厂创建的线程的 未捕获异常的行为
setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() { public void uncaughtException(Thread t,
Throwable e) {
log.log(Level.SEVERE, "UNCAUGHT in thread " + t.getName(), e);
}
});
} public void run() { // Copy debug flag to ensure consistent value throughout.
boolean debug = debugLifecycle; if (debug) log.log(Level.FINE, "Created " + getName()); try {
alive.incrementAndGet(); super.run();
} finally {
alive.decrementAndGet(); if (debug) log.log(Level.FINE, "Exiting " + getName());
}
} public static int getThreadsCreated() { return created.get();
} public static int getThreadsAlive() { return alive.get();
} public static boolean getDebug() { return debugLifecycle;
} public static void setDebug(boolean b) {
debugLifecycle = b;
}
}
5. カスタム サブクラスによってオーバーライドできるメソッド:
1.実行後:終了後、Run
timeExceptionがスローされた場合、メソッドは実行されません
2. beforeExecute: 開始前に、Runtime
Exceptionがスローされた場合、タスクは実行されません3.terminated: の場合スレッドプールは閉じられており、リソースの解放などに使用できます
2.再帰
アルゴリズムの並列化1.
ループ//串行化
void processSequentially(List<Element> elements) { for (Element e : elements)
process(e);
} //并行化
void processInParallel(Executor exec, List<Element> elements) { for (final Element e : elements)
exec.execute(new Runnable() { public void run() {
process(e);
}
});
}
2. 反復
各反復操作が互いに独立している場合、シリアル実行
例: 深さ優先探索
アルゴリズム 注: 再帰は依然としてシリアルです。 , ただし、各ノードの計算は並列です//串行 计算compute 和串行迭代
public <T> void sequentialRecursive(List<Node<T>> nodes, Collection<T> results) { for (Node<T> n : nodes) {
results.add(n.compute());
sequentialRecursive(n.getChildren(), results);
}
} //并行 计算compute 和串行迭代
public <T> void parallelRecursive(final Executor exec, List<Node<T>> nodes, final Collection<T> results) { for (final Node<T> n : nodes) {
exec.execute(() -> results.add(n.compute()));
parallelRecursive(exec, n.getChildren(), results);
}
} //调用并行方法的操作
public <T> Collection<T> getParallelResults(List<Node<T>> nodes) throws InterruptedException {
ExecutorService exec = Executors.newCachedThreadPool();
Queue<T> resultQueue = new ConcurrentLinkedQueue<T>();
parallelRecursive(exec, nodes, resultQueue);
exec.shutdown();
exec.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS); return resultQueue;
}
实例：
public class ConcurrentPuzzleSolver <P, M> { private final Puzzle<P, M> puzzle; private final ExecutorService exec; private final ConcurrentMap<P, Boolean> seen; protected final ValueLatch<PuzzleNode<P, M>> solution = new ValueLatch<PuzzleNode<P, M>>(); public ConcurrentPuzzleSolver(Puzzle<P, M> puzzle) { this.puzzle = puzzle; this.exec = initThreadPool(); this.seen = new ConcurrentHashMap<P, Boolean>(); if (exec instanceof ThreadPoolExecutor) {
ThreadPoolExecutor tpe = (ThreadPoolExecutor) exec;
tpe.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());
}
} private ExecutorService initThreadPool() { return Executors.newCachedThreadPool();
} public List<M> solve() throws InterruptedException { try {
P p = puzzle.initialPosition();
exec.execute(newTask(p, null, null)); // 等待ValueLatch中闭锁解开，则表示已经找到答案
PuzzleNode<P, M> solnPuzzleNode = solution.getValue(); return (solnPuzzleNode == null) ? null : solnPuzzleNode.asMoveList();
} finally {
exec.shutdown();//最终主线程关闭线程池 }
} protected Runnable newTask(P p, M m, PuzzleNode<P, M> n) { return new SolverTask(p, m, n);
} protected class SolverTask extends PuzzleNode<P, M> implements Runnable {
SolverTask(P pos, M move, PuzzleNode<P, M> prev) { super(pos, move, prev);
} public void run() { //如果有一个线程找到了答案，则return，通过ValueLatch中isSet CountDownlatch闭锁实现； //为类避免死锁，将已经扫描的节点放入set集合中，避免继续扫描产生死循环
if (solution.isSet() || seen.putIfAbsent(pos, true) != null){ return; // already solved or seen this position } if (puzzle.isGoal(pos)) {
solution.setValue(this);
} else { for (M m : puzzle.legalMoves(pos))
exec.execute(newTask(puzzle.move(pos, m), m, this));
}
}
}
}
以上がJavaスレッドプールの使用の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。