JAVA線程池學習以及隊列拒絕策略


工作中遇到了消息隊列的發送,之前都是用數據庫作為中轉和暫存的。這次考慮用多線程的方式進行消息的發送,於是學習了一下線程池的應用。說實話,實踐中對Java高級特性的應用真的不多,對多線程的理解也就一直停留在理論層面。借着實踐的機會好好整理一下。

准備從以下幾個方面總結:

  • 線程池的使用

  • 消息隊列——生產者消費者模式

  • 定時任務Quartz原理

  • 線程池的大小、隊列大小設置

這個部分是有關線程池的使用:

1. 為什么要用線程池?

在Java中,如果每當一個請求到達就創建一個新線程,開銷是相當大的。在實際使用中,每個請求創建新線程的服務器在創建和銷毀線程上花費的時間和消耗的系統資源,甚至可能要比花在實際處理實際的用戶請求的時間和資源要多的多。除了創建和銷毀線程的開銷之外,活動的線程也需要消耗系統資源。如果在一個JVM中創建太多的線程,可能會導致系統由於過度消耗內存或者“切換過度”而導致系統資源不足。為了防止資源不足,服務器應用程序需要一些辦法來限制任何給定時刻處理的請求數目,盡可能減少創建和銷毀線程的次數,特別是一些資源耗費比較大的線程的創建和銷毀,盡量利用已有對象來進行服務,這就是“池化資源”技術產生的原因。

線程池主要用來解決線程生命周期開銷問題和資源不足問題,通過對多個任務重用線程,線程創建的開銷被分攤到多個任務上了,而且由於在請求到達時線程已經存在,所以消除了創建所帶來的延遲。這樣,就可以立即請求服務,使應用程序響應更快。另外,通過適當的調整線程池中的線程數據可以防止出現資源不足的情況。

2. ThreadPoolExecutor類

JDK 1.5以后,Java提供一個線程池ThreadPoolExecutor類。下面從構造函數來分析一下這個線程池的使用方法。

public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler);

參數名、說明:

參數名 說明
corePoolSize 線程池維護線程的最少數量
maximumPoolSize 線程池維護線程的最大數量
keepAliveTime 線程池維護線程所允許的空閑時間
workQueue 任務隊列,用來存放我們所定義的任務處理線程
threadFactory 線程創建工廠
handler 線程池對拒絕任務的處理策略

ThreadPoolExecutor將根據corePoolSizemaximumPoolSize設置的邊界自動調整池大小。當新任務在方法execute(Runnable) 中提交時, 如果運行的線程少於corePoolSize,則創建新線程來處理請求。

如果正在運行的線程等於corePoolSize時,ThreadPoolExecutor優先往隊列中添加任務,直到隊列滿了,並且沒有空閑線程時才創建新的線程。如果設置的corePoolSize 和 maximumPoolSize 相同,則創建了固定大小的線程池。

keepAliveTime:當線程數達到maximumPoolSize時,經過某段時間,發現多出的線程出於空閑狀態,就進行線程的回收。keepAliveTime就是線程池內最大的空閑時間。

workQueue:當核心線程不能都在處理任務時,新進任務被放在Queue里。

線程池中任務有三種排隊策略:

a. 直接提交。直接提交策略表示線程池不對任務進行緩存。新進任務直接提交給線程池,當線程池中沒有空閑線程時,創建一個新的線程處理此任務。這種策略需要線程池具有無限增長的可能性。實現為:SynchronousQueue

b. 有界隊列。當線程池中線程達到corePoolSize時,新進任務被放在隊列里排隊等待處理。有界隊列(如ArrayBlockingQueue)有助於防止資源耗盡,但是可能較難調整和控制。隊列大小和最大池大小可能需要相互折衷:使用大型隊列和小型池可以最大限度地降低 CPU 使用率、操作系統資源和上下文切換開銷,但是可能導致人工降低吞吐量。如果任務頻繁阻塞(例如,如果它們是 I/O 邊界),則系統可能為超過您許可的更多線程安排時間。使用小型隊列通常要求較大的池大小,CPU 使用率較高,但是可能遇到不可接受的調度開銷,這樣也會降低吞吐量。

c. 無界隊列。使用無界隊列(例如,不具有預定義容量的 LinkedBlockingQueue)將導致在所有 corePoolSize 線程都忙時新任務在隊列中等待。這樣,創建的線程就不會超過 corePoolSize。(因此,maximumPoolSize 的值也就無效了。)當每個任務完全獨立於其他任務,即任務執行互不影響時,適合於使用無界隊列;例如,在 Web 頁服務器中。這種排隊可用於處理瞬態突發請求,當命令以超過隊列所能處理的平均數連續到達時,此策略允許無界線程具有增長的可能性。

拒絕策略:當任務源源不斷的過來,而我們的系統又處理不過來的時候,我們要采取的策略是拒絕服務。RejectedExecutionHandler接口提供了拒絕任務處理的自定義方法的機會。在ThreadPoolExecutor中已經包含四種處理策略。

1)CallerRunsPolicy:線程調用運行該任務的 execute 本身。此策略提供簡單的反饋控制機制,能夠減緩新任務的提交速度。

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}

這個策略顯然不想放棄執行任務。但是由於池中已經沒有任何資源了,那么就直接使用調用該execute的線程本身來執行。(開始我總不想丟棄任務的執行,但是對某些應用場景來講,很有可能造成當前線程也被阻塞。如果所有線程都是不能執行的,很可能導致程序沒法繼續跑了。需要視業務情景而定吧。)

2)AbortPolicy:處理程序遭到拒絕將拋出運行時 RejectedExecutionException

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException();
}

這種策略直接拋出異常,丟棄任務。(jdk默認策略,隊列滿並線程滿時直接拒絕添加新任務,並拋出異常,所以說有時候放棄也是一種勇氣,為了保證后續任務的正常進行,丟棄一些也是可以接收的,記得做好記錄)

3)DiscardPolicy:不能執行的任務將被刪除

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {}

這種策略和AbortPolicy幾乎一樣,也是丟棄任務,只不過他不拋出異常。

4)DiscardOldestPolicy:如果執行程序尚未關閉,則位於工作隊列頭部的任務將被刪除,然后重試執行程序(如果再次失敗,則重復此過程)

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}

該策略就稍微復雜一些,在pool沒有關閉的前提下首先丟掉緩存在隊列中的最早的任務,然后重新嘗試運行該任務。這個策略需要適當小心。

3. Executors 工廠

ThreadPoolExecutor是Executors類的實現,Executors類里面提供了一些靜態工廠,生成一些常用的線程池,主要有以下幾個:

newSingleThreadExecutor:創建一個單線程的線程池。這個線程池只有一個線程在工作,也就是相當於單線程串行執行所有任務。如果這個唯一的線程因為異常結束,那么會有一個新的線程來替代它。此線程池保證所有任務的執行順序按照任務的提交順序執行。

newFixedThreadPool:創建固定大小的線程池。每次提交一個任務就創建一個線程,直到線程達到線程池的最大大小。線程池的大小一旦達到最大值就會保持不變,如果某個線程因為執行異常而結束,那么線程池會補充一個新線程。(我用的就是這個,同上所述,相當於創建了相同corePoolSize、maximumPoolSize的線程池)

newCachedThreadPool:創建一個可緩存的線程池。如果線程池的大小超過了處理任務所需要的線程,那么就會回收部分空閑(60秒不執行任務)的線程,當任務數增加時,此線程池又可以智能的添加新線程來處理任務。此線程池不會對線程池大小做限制,線程池大小完全依賴於操作系統(或者說JVM)能夠創建的最大線程大小。

4. 舉個栗子

測試類:ThreadPool,創建了一個 newFixedThreadPool,最大線程數為3的固定大小線程池。然后模擬10個任務丟進去。主線程結束后會打印一句:主線程結束。

public class ThreadPool {
public static void main(String[] args) {
//創建固定大小線程池
ExecutorService executor = Executors.newFixedThreadPool(3);
for (int i = 0; i < 10; i++) {
SendNoticeTask task = new SendNoticeTask();
task.setCount(i);
executor.execute(task);
}
System.out.println("主線程結束");
}
}

測試類:SendNoticeTask,執行任務類,就是打印一句當前線程名+第幾個任務。為了方便觀察,每個線程執行完以后睡10s。

public class SendNoticeTask implements Runnable {

private int count;

public void setCount(int count) {
this.count = count;
}

@Override
public void run() {
System.out.println(Thread.currentThread().getName() + " start to" + " send " + count + " ...");
try {
Thread.currentThread().sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("finish " + Thread.currentThread().getName());
}
}

執行結果:

主線程結束
pool-1-thread-3 start to send 2 ...
pool-1-thread-1 start to send 0 ...
pool-1-thread-2 start to send 1 ...
finish pool-1-thread-3
finish pool-1-thread-2
pool-1-thread-2 start to send 3 ...
finish pool-1-thread-1
pool-1-thread-3 start to send 4 ...
pool-1-thread-1 start to send 5 ...
finish pool-1-thread-3
finish pool-1-thread-1
pool-1-thread-1 start to send 6 ...
pool-1-thread-3 start to send 7 ...
finish pool-1-thread-2
pool-1-thread-2 start to send 8 ...
finish pool-1-thread-1
pool-1-thread-1 start to send 9 ...
finish pool-1-thread-3
finish pool-1-thread-2
finish pool-1-thread-1

由上可見執行順序是這樣的:線程池創建了三個線程,分別執行任務0、1、2,由於線程創建需要一定時間,因此前三個線程的執行順序具有一定隨機性。此時主線程接着往線程池中塞任務,線程池已達到最大線程數(3),於是開始往隊列里放。當某個線程執行完任務后,直接從隊列里拉出新的任務執行,隊列具有先進先出的特性,因此后面的任務執行是有序的。

這個看一下 Executors 類的源碼就更明白了。

public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}

實際上是創建了一個具有固定線程數、無界隊列的 ThreadPoolExecutor。隊列無界,不會拒絕任務提交,因此使用此方法時,需要注意資源被耗盡的情況。

測試類:TestThreadPool,采用有界隊列(隊列大小2),和默認拒絕策略的ThreadPoolExecutor。

public class TestThreadPool {

private static final int corePoolSize = 2;
private static final int maximumPoolSize = 4;
private static final int keepAliveTime = 1000;
private static BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<Runnable>(2);

public static void main(String[] args) {
//創建線程池
ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime,
TimeUnit.MILLISECONDS, workQueue);
for (int i = 0; i < 10; i++) {
SendNoticeTask task = new SendNoticeTask();
task.setCount(i);
executor.execute(task);
}
System.out.println("主線程結束:" + Thread.currentThread().getName());
}
}

執行結果:

pool-1-thread-1 start to send 0 ...
Exception in thread "main" java.util.concurrent.RejectedExecutionException
at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:1768)
at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:767)
at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:658)
at com.dylanviviv.pool.TestThreadPool.main(TestThreadPool.java:24)
pool-1-thread-4 start to send 5 ...
pool-1-thread-2 start to send 1 ...
pool-1-thread-3 start to send 4 ...
finish pool-1-thread-1
finish pool-1-thread-3
pool-1-thread-1 start to send 2 ...
finish pool-1-thread-2
pool-1-thread-3 start to send 3 ...
finish pool-1-thread-4
finish pool-1-thread-3
finish pool-1-thread-1

線程池創建了2個線程,分別執行任務0、1,線程池達到corePoolSize,新進任務2、3被放入隊列中等待處理,此時隊列滿,而線程池中線程沒有執行完任務0、1,線程池創建新的線程,執行新進任務4、5,達到maximumPoolSize。此時所有任務都沒有執行結束,主線程又繼續提交任務,線程池進入默認異常策略(AbortPolicy)拒絕服務。


注意!

本站转载的文章为个人学习借鉴使用,本站对版权不负任何法律责任。如果侵犯了您的隐私权益,请联系我们删除。



 
粤ICP备14056181号  © 2014-2020 ITdaan.com