接系列3，在該系列中我們一起探討一下CachedThreadPool。
線程池是Conncurrent包提供給我們的一個重要的禮物。使得我們沒有必要維護自個實現(xiàn)的心里很沒底的線程池了。但如何充分利用好這些線程池來加快我們開發(fā)與測試效率呢？當然是知己知彼。本系列就說說對CachedThreadPool使用的一下問題。
下面是對CachedThreadPool的一個測試，程序有問題嗎？
package net.blogjava.vincent;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class CachedThreadPoolIssue {
/**
* @param args
*/
public static void main(String[] args) {
ExecutorService es = Executors.newCachedThreadPool();
for(int i = 1; i<8000; i++)
es.submit(new task());
}
}
class task implements Runnable{
@Override
public void run() {
try {
Thread.sleep(4000);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
如果對JVM沒有特殊的設置，并在Window平臺上，那么就會有一下異常的發(fā)生：
Exception in thread "main" java.lang.OutOfMemoryError: unable to create new native thread
at java.lang.Thread.start0(Native Method)
at java.lang.Thread.start(Unknown Source)
at java.util.concurrent.ThreadPoolExecutor.addIfUnderMaximumPoolSize(Unknown Source)
at java.util.concurrent.ThreadPoolExecutor.execute(Unknown Source)
at java.util.concurrent.AbstractExecutorService.submit(Unknown Source)
at net.blogjava.vincent.CachedThreadPoolIssue.main(CachedThreadPoolIssue.java:19)
看看Doc對該線程池的介紹：
Creates a thread pool that creates new threads as needed, but will reuse previously constructed threads when they are available. These pools will typically improve the performance of programs that execute many short-lived asynchronous tasks. Calls to execute will reuse previously constructed threads if available. If no existing thread is available, a new thread will be created and added to the pool. Threads that have not been used for sixty seconds are terminated and removed from the cache. Thus, a pool that remains idle for long enough will not consume any resources. Note that pools with similar properties but different details (for example, timeout parameters) may be created using ThreadPoolExecutor constructors.
有以下幾點需要注意：
1. 指出會重用先前的線程，不錯。
2. 提高了短Task的吞吐量。
3. 線程如果60s沒有使用就會移除出Cache。
好像跟剛才的錯誤沒有關系，其實就第一句話說了問題，它會按需要創(chuàng)建新的線程，上面的例子一下提交8000個Task，意味著該線程池就會創(chuàng)建8000線程，當然，這遠遠高于JVM限制了。
注：在JDK1.5中，默認每個線程使用1M內存，8000M !!! 可能嗎??！
所以我感覺這應該是我遇到的第一個Concurrent不足之處，既然這么設計，那么就應該在中Doc指出，應該在使用避免大量Task提交到給CachedThreadPool.
可能讀者不相信，那么下面的例子說明了他創(chuàng)建的Thread。
在ThreadPoolExecutor提供的API中，看到它提供beforeExecute 和afterExecute兩個可以在子類中重載的方法，該方法在線程池中線程執(zhí)行Task之前與之后調用。所以我們在beforeExexute中查看目前線程編號就可以確定目前的線程數(shù)目.
package net.blogjava.vincent;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class CachedThreadPoolIssue {
/**
* @param args
*/
public static void main(String[] args) {
ExecutorService es = new LogThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
for(int i = 1; i<8000; i++)
es.submit(new task());
}
}
class task implements Runnable{
@Override
public void run() {
try {
Thread.sleep(600000);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
class LogThreadPoolExecutor extends ThreadPoolExecutor{
public LogThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
}
protected void beforeExecute(Thread t, Runnable r) {
System.out.println(t.getName());
}
protected void afterExecute(Runnable r, Throwable t) {
}
}
測試結果如圖：
當線程數(shù)達到5592是，只有在任務管理器Kill該進程了。
如何解決該問題呢，其實在剛才實例化時就看出來了，只需將
new LogThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
Integer.MAX_VALUE 改為合適的大小。對于該參數(shù)的含義，涉及到線程池的實現(xiàn)，將會在下個系列中指出。
當然，其他的解決方案就是控制Task的提交速率，避免超過其最大限制。