《Java併發程式設計實戰》筆記3——執行緒池的使用
1、執行緒飢餓死鎖
線上程池中,如果任務依賴於其他任務,那麼可能發生死鎖。在單執行緒的Executor中,如果一個任務將另一個任務提交到同一個Executor,並且等待這個被提交任務的結果,那麼通常會引發死鎖。
如下面程式碼所示:
public class ThreadDeadlock { ExecutorService exec = Executors.newSingleThreadExecutor();//使用單執行緒的Executor public class LoadFileTask implements Callable<String> { private final String fileName; public LoadFileTask(String fileName) { this.fileName = fileName; } public String call() throws Exception { // Here's where we would actually read the file return ""; } } public class RenderPageTask implements Callable<String> { public String call() throws Exception { Future<String> header, footer; header = exec.submit(new LoadFileTask("header.html")); footer = exec.submit(new LoadFileTask("footer.html")); String page = renderBody(); System.out.println("提交子任務"); // Will deadlock -- task waiting for result of subtask return header.get() + page + footer.get(); } private String renderBody() { // Here's where we would actually render the page return ""; } } public static void main(String[] args) { ThreadDeadlock deadLock = new ThreadDeadlock(); ThreadDeadlock.RenderPageTask t = deadLock.new RenderPageTask(); Future<String> result = deadLock.exec.submit(t); System.out.println("提交主任務"); try { System.out.println("獲取結果"); result.get(); System.out.println("===結束==="); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } } }
執行結果:
2、設定執行緒池的大小
如果執行緒池過大,大量的執行緒將在相對較少的CUP和記憶體資源上發生競爭,這不僅會導致更高的記憶體使用量,而且還可能耗盡資源。如果執行緒池過小,那麼將導致許多空閒的處理器無法執行工作,從而降低吞吐率。
3、配置ThreadPoolExecutor
(1)引數簡介
ThreadPoolExecutor實現了ExecutorService和Executors介面,通常可通過Executors
工廠方法來配置引數。較為完整的構造器如下所示。
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit timeUnit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler)
corePoolSize:核心執行緒數(執行緒池努力去保持的一個執行緒數量)
maximumPoolSize:最大執行緒數
keepAliveTime:執行緒空閒時間。當執行緒數>corePoolSize,空閒執行緒最多等待keepAliveTime時間,便會銷燬。
timeUnit :keepAliveTime的時間單位
workQueue:任務佇列,儲存等待執行的任務。
threadFactory :用來建立新執行緒的工廠
rejectedExecutionHandler :任務拒絕處理器。當拒絕任務時會觸發
執行順序:
①執行緒數<corePoolSize,即使有其他空閒的工作執行緒,也會建立一個新執行緒來處理請求。
②如果執行緒數>=corePoolSize,會將新任務放在佇列中等待
③如果佇列滿了,且執行緒數<maximumPoolSize,才會建立一個新執行緒來處理請求。
④如果佇列滿了,且執行緒數>=maximumPoolSize,拒絕請求
如果將corePoolSize與maximumPoolSize設定成一樣,那麼會的到一個固定大小的執行緒池。
通過設定maximumPoolSize為一個無限大的數值(如Integer.MAX_VALUE),該執行緒池能適應任意數量的併發任務。
(2)管理佇列任務
ThreadPoolExecutor可以提供一個BlockingQueue來儲存等待執行的任務。基本的任務排隊方法有三種:無界佇列、有界佇列和同步移交。
只有當任務相互獨立時,為執行緒池或工作佇列設定界限才是合理的。如果任務之間存在依賴性,那麼有界的執行緒池或佇列就可能導致執行緒“飢餓”死鎖問題。此時應該使用無界的執行緒池,如:newCachedThreadPool。
(3)任務拒絕處理策略(RejectedExecutionHandler )
當任務新增到執行緒池中被拒絕時:
①ThreadPoolExecutor.AbortPolicy。會丟擲RejectedExecutionException異常
②ThreadPoolExecutor.CallerRunsPolicy。將某些任務回退到呼叫者,從而降低新任務的流量。它不會線上程池的某個執行緒中執行新提交的任務,而是在一個呼叫了execute的執行緒中執行該任務。
③ThreadPoolExecutor.DiscardPolicy。無法執行的任務會被簡單地拋棄
④ThreadPoolExecutor.DiscardOldestPolicy。如果executor沒有關閉,在任務佇列頭的任務會被拋棄,然後會再次嘗試重新提交新的任務(可能會再次失敗,繼續重複這個操作)
(4)執行緒工廠(ThreadFactory )
執行緒池需要建立新的執行緒時,都是通過執行緒工廠方法來完成的。但有時需要使用定製的執行緒工廠方法。
public interface ThreadFactory {
Thread newThread(Runnable r);
}
/**
* 自定義執行緒工廠
*/
public class MyThreadFactory implements ThreadFactory {
private final String poolName;//執行緒池名
public MyThreadFactory(String poolName) {
this.poolName = poolName;
}
@Override
public Thread newThread(Runnable runnable) {
return new MyAppThread(runnable, poolName);
}
}
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());
//設定日誌顯示級別
log.setLevel(Level.FINE);
//Set the handler invoked when this thread abruptly terminates due to an uncaught exception
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.SEVERE, "Created " + getName());
try {
alive.incrementAndGet();
super.run();
} finally {
alive.decrementAndGet();
if (debug) log.log(Level.SEVERE, "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;
}
}
測試:
public class TestMain {
public static void main(String[] args) {
MyThreadFactory factory = new MyThreadFactory("Wow");
MyAppThread.setDebug(true);
MyAppThread t =(MyAppThread) factory.newThread(new Runnable() {
@Override
public void run() {
int a = 1/0;//沒有捕獲的異常
}
});
t.start();
//*******************************************************************//
MyThreadFactory factory2 = new MyThreadFactory("hoo");
MyAppThread.setDebug(true);
MyAppThread t2 =(MyAppThread) factory2.newThread(new Runnable() {
@Override
public void run() {
int a = 1/0;//沒有捕獲的異常
}
});
t2.start();
}
}
測試結果:
(5)擴充套件ThreadPoolExecutor
可以根據需求擴充套件ThreadPoolExecutor重寫beforeExecute、afterExecute、terminated等方法