溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
今天就跟大家聊聊有關在java項目中使用線程池實現(xiàn)并發(fā)編程，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結了以下內(nèi)容，希望大家根據(jù)這篇文章可以有所收獲。
一、任務和執(zhí)行策略之間的隱性耦合
Executor可以將任務的提交和任務的執(zhí)行策略解耦
只有任務是同類型的且執(zhí)行時間差別不大，才能發(fā)揮最大性能，否則，如將一些耗時長的任務和耗時短的任務放在一個線程池，除非線程池很大，否則會造成死鎖等問題
1.線程饑餓死鎖
類似于：將兩個任務提交給一個單線程池，且兩個任務之間相互依賴，一個任務等待另一個任務，則會發(fā)生死鎖；表現(xiàn)為池不夠
定義：某個任務必須等待池中其他任務的運行結果，有可能發(fā)生饑餓死鎖
2.線程池大小
注意：線程池的大小還受其他的限制，如其他資源池：數(shù)據(jù)庫連接池
如果每個任務都是一個連接，那么線程池的大小就受制于數(shù)據(jù)庫連接池的大小
3.配置ThreadPoolExecutor線程池
實例：
1.通過Executors的工廠方法返回默認的一些實現(xiàn)
2.通過實例化ThreadPoolExecutor（.....）自定義實現(xiàn)
線程池的隊列
1.無界隊列：任務到達，線程池飽滿，則任務在隊列中等待，如果任務無限達到，則隊列會無限擴張
如：單例和固定大小的線程池用的就是此種
2.有界隊列：如果新任務到達，隊列滿則使用飽和策略
3.同步移交：如果線程池很大，將任務放入隊列后在移交就會產(chǎn)生延時，如果任務生產(chǎn)者很快也會導致任務排隊
SynchronousQueue直接將任務移交給工作線程
機制：將一個任務放入，必須有一個線程等待接受，如果沒有，則新增線程，如果線程飽和，則拒絕任務
如：CacheThreadPool就是使用的這種策略
飽和策略：
setRejectedExecutionHandler來修改飽和策略
1.終止Abort（默認）：拋出異常由調(diào)用者處理
2.拋棄Discard
3.拋棄DiscardOldest：拋棄最舊的任務，注意：如果是優(yōu)先級隊列將拋棄優(yōu)先級最高的任務
4.CallerRuns：回退任務，有調(diào)用者線程自行處理
4.線程工廠ThreadFactoy
每當創(chuàng)建線程時：其實是調(diào)用了線程工廠來完成
自定義線程工廠：implements ThreadFactory
可以定制該線程工廠的行為：如UncaughtExceptionHandler等
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());
//設置該線程工廠創(chuàng)建的線程的 未捕獲異常的行為
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.擴展ThreadPoolExecutor
可以被自定義子類覆蓋的方法：
1.afterExecute：結束后，如果拋出RuntimeException則方法不會執(zhí)行
2.beforeExecute：開始前，如果拋出RuntimeException則任務不會執(zhí)行
3.terminated：在線程池關閉時，可以用來釋放資源等
二、遞歸算法的并行化
1.循環(huán)
在循環(huán)中，每次循環(huán)操作都是獨立的
//串行化
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.迭代
如果每個迭代操作是彼此獨立的，則可以串行執(zhí)行
如：深度優(yōu)先搜索算法；注意：遞歸還是串行的，但是，每個節(jié)點的計算是并行的
//串行 計算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);
}
}
//調(diào)用并行方法的操作
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中閉鎖解開，則表示已經(jīng)找到答案
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閉鎖實現(xiàn)；
//為類避免死鎖，將已經(jīng)掃描的節(jié)點放入set集合中，避免繼續(xù)掃描產(chǎn)生死循環(huán)
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));
}
}
}
}看完上述內(nèi)容，你們對在java項目中使用線程池實現(xiàn)并發(fā)編程有進一步的了解嗎？如果還想了解更多知識或者相關內(nèi)容，請關注億速云行業(yè)資訊頻道，感謝大家的支持。
免責聲明：本站發(fā)布的內(nèi)容（圖片、視頻和文字）以原創(chuàng)、轉載和分享為主，文章觀點不代表本網(wǎng)站立場，如果涉及侵權請聯(lián)系站長郵箱：is@yisu.com進行舉報，并提供相關證據(jù)，一經(jīng)查實，將立刻刪除涉嫌侵權內(nèi)容。
