Java 線程池框架
new Thread的弊端
執行一個異步任務你還只是以下new Thread嗎？數據庫
new Thread(new Runnable() {
@Override
public void run() {
// TODO Auto-generated method stub
}
}).start();
那你就out太多了，new Thread的弊端以下：緩存
- 每次new Thread新建對象性能差。
- 線程缺少統一管理，可能無限制新建線程，相互之間競爭，及可能佔用過多系統資源致使死機或oom。
- 缺少更多功能，如定時執行、按期執行、線程中斷。
相比new Thread，Java提供的四種線程池的好處在於：安全
- 重用存在的線程，減小對象建立、消亡的開銷，性能佳。
- 可有效控制最大併發線程數，提升系統資源的使用率，同時避免過多資源競爭，避免堵塞。
- 提供定時執行、按期執行、單線程、併發數控制等功能。
Java 線程池
Java經過Executors提供四種線程池，分別爲：併發
- newCachedThreadPool：建立一個可緩存線程池，若是線程池長度超過處理須要，可靈活回收空閒線程，若無可回收，則新建線程。
- newFixedThreadPool：建立一個定長線程池，可控制線程最大併發數，超出的線程會在隊列中等待。
- newScheduledThreadPool：建立一個定長線程池，支持定時及週期性任務執行。
- newSingleThreadExecutor：建立一個單線程化的線程池，它只會用惟一的工做線程來執行任務，保證全部任務按照指定順序(FIFO, LIFO, 優先級)執行。
newCachedThreadPool
建立一個可緩存線程池，若是線程池長度超過處理須要，可靈活回收空閒線程，若無可回收，則新建線程。示例代碼以下：異步
ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
final int index = i;
try {
Thread.sleep(index * 1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
cachedThreadPool.execute(new Runnable() {
@Override
public void run() {
System.out.println(index);
}
});
}
線程池爲無限大，當執行第二個任務時第一個任務已經完成，會複用執行第一個任務的線程，而不用每次新建線程。ide
newFixedThreadPool
建立一個定長線程池，可控制線程最大併發數，超出的線程會在隊列中等待。示例代碼以下：函數
ExecutorService fixedThreadPool = Executors.newFixedThreadPool(3);
for (int i = 0; i < 10; i++) {
final int index = i;
fixedThreadPool.execute(new Runnable() {
@Override
public void run() {
try {
System.out.println(index);
Thread.sleep(2000);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
});
}
由於線程池大小爲3，每一個任務輸出index後sleep 2秒，因此每兩秒打印3個數字。 定長線程池的大小最好根據系統資源進行設置。如Runtime.getRuntime().availableProcessors()。可參考PreloadDataCache。性能
newScheduledThreadPool
建立一個定長線程池，支持定時及週期性任務執行。延遲執行示例代碼以下：this
ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);
scheduledThreadPool.schedule(new Runnable() {
@Override
public void run() {
System.out.println("delay 3 seconds");
}
}, 3, TimeUnit.SECONDS);
表示延遲3秒執行。線程
按期執行示例代碼以下：
scheduledThreadPool.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
System.out.println("delay 1 seconds, and excute every 3 seconds");
}
}, 1, 3, TimeUnit.SECONDS);
表示延遲1秒後每3秒執行一次。 ScheduledExecutorService比Timer更安全，功能更強大，後面會有一篇單獨進行對比。
newSingleThreadExecutor
建立一個單線程化的線程池，它只會用惟一的工做線程來執行任務，保證全部任務按照指定順序(FIFO, LIFO, 優先級)執行。示例代碼以下：
ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
for (int i = 0; i < 10; i++) {
final int index = i;
singleThreadExecutor.execute(new Runnable() {
@Override
public void run() {
try {
System.out.println(index);
Thread.sleep(2000);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
});
}
結果依次輸出，至關於順序執行各個任務。 現行大多數GUI程序都是單線程的。Android中單線程可用於數據庫操做，文件操做，應用批量安裝，應用批量刪除等不適合併發但可能IO阻塞性及影響UI線程響應的操做。
CompletionService
當使用ExecutorService啓動了多個Callable後，每一個Callable會產生一個Future，咱們須要將多個Future存入一個線性表，用於以後處理數據。固然，還有更復雜的狀況，有5個生產者線程，每一個生產者線程都會建立任務，全部任務的Future都存放到同一個線性表中。而後遍歷線性表，經過調用future.get(0, TimeUnit.SECONDS)不斷嘗試獲取完成結果，直到獲取到全部的結果。以下
public class ExecutorServiceTest {
static class Task implements Callable<String>{
private int i;
public Task(int i){
this.i = i;
}
@Override
public String call() throws Exception {
Thread.sleep(10000);
return Thread.currentThread().getName() + "執行完任務：" + i;
}
}
public static void main(String[] args){
testUseFuture();
}
private static void testUseFuture(){
int numThread = 5;
ExecutorService executor = Executors.newFixedThreadPool(numThread);
List<Future<String>> futureList = new ArrayList<Future<String>>();
for(int i = 0;i<numThread;i++ ){
Future<String> future = executor.submit(new ExecutorServiceTest.Task(i));
futureList.add(future);
}
int i=0;
while(numThread > 0){
i++;
for(Future<String> future : futureList){
String result = null;
try {
result = future.get(0, TimeUnit.SECONDS);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
} catch (TimeoutException e) {
//超時異常直接忽略
}
if(null != result){
futureList.remove(future);
numThread--;
System.out.println(result);
//此處必須break，不然會拋出併發修改異常。（也能夠經過將futureList聲明爲CopyOnWriteArrayList類型解決）
break;
}
}
}
System.out.println("共"+i+"次遍歷任務列表");
}
}
執行後輸出
pool-1-thread-2執行完任務：1 pool-1-thread-3執行完任務：2 pool-1-thread-4執行完任務：3 pool-1-thread-1執行完任務：0 pool-1-thread-5執行完任務：4 共1265559次遍歷任務列表
根據輸出咱們能夠看到，爲了獲取到5個任務的執行結果，程序一共遍歷了1265559次任務列表，這不是一種很是好的辦法。
CompletionService正是爲此而存在，它是一個更高級的ExecutorService，它自己自帶一個線程安全的線性表，無需用戶額外建立。它提供了3種方法從線性表中取出結果：poll()是非阻塞的，若目前無結果，返回一個null，線程繼續運行不阻塞；poll(long timeout, TimeUnit unit)是阻塞的，若目前無結果，則會等待一段時間；**take()**是阻塞的，若當前無結果，則線程阻塞，直到產生一個結果，被取出返回，線程才繼續運行。 修改後的程序以下：
public class CompletionServiceTest {
static class Task implements Callable<String>{
private int i;
public Task(int i){
this.i = i;
}
@Override
public String call() throws Exception {
Thread.sleep(10000);
return Thread.currentThread().getName() + "執行完任務：" + i;
}
}
public static void main(String[] args) throws InterruptedException, ExecutionException{
testExecutorCompletionService();
}
private static void testExecutorCompletionService() throws InterruptedException, ExecutionException{
int numThread = 5;
ExecutorService executor = Executors.newFixedThreadPool(numThread);
CompletionService<String> completionService = new ExecutorCompletionService<String>(executor);
for(int i = 0;i<numThread;i++ ){
completionService.submit(new CompletionServiceTest.Task(i));
}
for(int i = 0;i<numThread;i++ ){
System.out.println(completionService.take().get());
System.out.println("第"+(i+1)+"次獲取結果");
}
}
}
使用completionService.take()阻塞方法來獲取已完成Future<>，不須要一直遍歷查詢。
pool-1-thread-2執行完任務：1 第1次獲取結果 pool-1-thread-3執行完任務：2 第2次獲取結果 pool-1-thread-1執行完任務：0 第3次獲取結果 pool-1-thread-5執行完任務：4 第4次獲取結果 pool-1-thread-4執行完任務：3 第5次獲取結果
CompletionService整合了Executor和BlockingQueue的功能。你能夠將Callable任務提交給它去執行，而後使用相似於隊列中的take和poll方法，在結果完整可用時得到這個結果，像一個打包的Future。ExecutorCompletionService是實現CompletionService接口的一個類，並將計算任務委託給一個Executor。
ExecutorCompletionService的實現至關直觀。它在構造函數中建立一個BlockingQueue，用它去保持完成的結果。計算完成時會調用FutureTask中的done方法。當提交一個任務後，首先把這個任務包裝爲一個QueueingFuture，它是FutureTask的一個子類，而後覆寫done方法，將結果置入BlockingQueue中，take和poll方法委託給了BlockingQueue，它會在結果不可用時阻塞。
public class ExecutorCompletionService<V> implements CompletionService<V> {
private final Executor executor;
private final AbstractExecutorService aes;
private final BlockingQueue<Future<V>> completionQueue;
/**
* FutureTask extension to enqueue upon completion
*/
private class QueueingFuture extends FutureTask<Void> {
QueueingFuture(RunnableFuture<V> task) {
super(task, null);
this.task = task;
}
protected void done() { completionQueue.add(task); }
private final Future<V> task;
}
private RunnableFuture<V> newTaskFor(Callable<V> task) {
if (aes == null)
return new FutureTask<V>(task);
else
return aes.newTaskFor(task);
}
private RunnableFuture<V> newTaskFor(Runnable task, V result) {
if (aes == null)
return new FutureTask<V>(task, result);
else
return aes.newTaskFor(task, result);
}
/**
* Creates an ExecutorCompletionService using the supplied
* executor for base task execution and a
* {@link LinkedBlockingQueue} as a completion queue.
*
* @param executor the executor to use
* @throws NullPointerException if executor is <tt>null</tt>
*/
public ExecutorCompletionService(Executor executor) {
if (executor == null)
throw new NullPointerException();
this.executor = executor;
this.aes = (executor instanceof AbstractExecutorService) ?
(AbstractExecutorService) executor : null;
this.completionQueue = new LinkedBlockingQueue<Future<V>>();
}
/**
* Creates an ExecutorCompletionService using the supplied
* executor for base task execution and the supplied queue as its
* completion queue.
*
* @param executor the executor to use
* @param completionQueue the queue to use as the completion queue
* normally one dedicated for use by this service
* @throws NullPointerException if executor or completionQueue are <tt>null</tt>
*/
public ExecutorCompletionService(Executor executor,
BlockingQueue<Future<V>> completionQueue) {
if (executor == null || completionQueue == null)
throw new NullPointerException();
this.executor = executor;
this.aes = (executor instanceof AbstractExecutorService) ?
(AbstractExecutorService) executor : null;
this.completionQueue = completionQueue;
}
public Future<V> submit(Callable<V> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<V> f = newTaskFor(task);
executor.execute(new QueueingFuture(f));
return f;
}
public Future<V> submit(Runnable task, V result) {
if (task == null) throw new NullPointerException();
RunnableFuture<V> f = newTaskFor(task, result);
executor.execute(new QueueingFuture(f));
return f;
}
public Future<V> take() throws InterruptedException {
return completionQueue.take();
}
public Future<V> poll() {
return completionQueue.poll();
}
public Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException {
return completionQueue.poll(timeout, unit);
}
}
相關文章
- 1. Java線程池Executor框架
- 2. 線程池與Executor 框架
- 3. 線程池之Executor框架
- 4. Java併發——線程池Executor框架
- 5. Java中的線程池——Executor框架
- 6. 線程 線程池 Executor框架 讀筆
- 7. LINUX c++線程池框架
- 8. Linux C++線程池框架
- 9. 線程池（Executor框架）
- 10. 線程池的好處&線程池框架Executor
- 更多相關文章...
- • C# 多線程 - C#教程
- • Java Spring框架是什麼？它有哪些好處？ - Spring教程
- • Java 8 Stream 教程
- • YAML 入門教程
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
相關文章