線程池ThreadPoolExecutor使用簡介

時間 2020-02-12

標籤線程 threadpoolexecutor 使用簡介简体版

原文原文鏈接

1、簡介
線程池類爲 java.util.concurrent.ThreadPoolExecutor，經常使用構造方法爲：

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
long keepAliveTime, TimeUnit unit,
BlockingQueue workQueue,
RejectedExecutionHandler handler)
corePoolSize：線程池維護線程的最少數量
maximumPoolSize：線程池維護線程的最大數量
keepAliveTime：線程池維護線程所容許的空閒時間
unit：線程池維護線程所容許的空閒時間的單位
workQueue：線程池所使用的緩衝隊列
handler：線程池對拒絕任務的處理策略

一個任務經過 execute(Runnable)方法被添加到線程池，任務就是一個 Runnable類型的對象，任務的執行方法就是 Runnable類型對象的run()方法。

當一個任務經過execute(Runnable)方法欲添加到線程池時：

若是此時線程池中的數量小於corePoolSize，即便線程池中的線程都處於空閒狀態，也要建立新的線程來處理被添加的任務。
若是此時線程池中的數量等於 corePoolSize，可是緩衝隊列 workQueue未滿，那麼任務被放入緩衝隊列。
若是此時線程池中的數量大於corePoolSize，緩衝隊列workQueue滿，而且線程池中的數量小於maximumPoolSize，建新的線程來處理被添加的任務。
若是此時線程池中的數量大於corePoolSize，緩衝隊列workQueue滿，而且線程池中的數量等於maximumPoolSize，那麼經過 handler所指定的策略來處理此任務。

也就是：處理任務的優先級爲：
核心線程corePoolSize、任務隊列workQueue、最大線程maximumPoolSize，若是三者都滿了，使用handler處理被拒絕的任務。

當線程池中的線程數量大於 corePoolSize時，若是某線程空閒時間超過keepAliveTime，線程將被終止。這樣，線程池能夠動態的調整池中的線程數。

unit可選的參數爲java.util.concurrent.TimeUnit中的幾個靜態屬性：
NANOSECONDS、MICROSECONDS、MILLISECONDS、SECONDS。

workQueue我經常使用的是：java.util.concurrent.ArrayBlockingQueue

handler有四個選擇：
ThreadPoolExecutor.AbortPolicy()
拋出java.util.concurrent.RejectedExecutionException異常
ThreadPoolExecutor.CallerRunsPolicy()
重試添加當前的任務，他會自動重複調用execute()方法
ThreadPoolExecutor.DiscardOldestPolicy()
拋棄舊的任務
ThreadPoolExecutor.DiscardPolicy()
拋棄當前的任務

2、通常用法舉例

點擊(此處)摺疊或打開 java

package demo;
import java.io.Serializable;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class TestThreadPool2
{
private static int produceTaskSleepTime = 2;
private static int produceTaskMaxNumber = 10;
public static void main(String[] args)
{
// 構造一個線程池
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 4, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(3),
new ThreadPoolExecutor.DiscardOldestPolicy());
for (int i = 1; i <= produceTaskMaxNumber; i++)
{
try
{
// 產生一個任務，並將其加入到線程池
String task = "task@ " + i;
System.out.println("put " + task);
threadPool.execute(new ThreadPoolTask(task));
// 便於觀察，等待一段時間
Thread.sleep(produceTaskSleepTime);
}
catch (Exception e)
{
e.printStackTrace();
}
}
}
}
/**
* 線程池執行的任務
*/
class ThreadPoolTask implements Runnable, Serializable
{
private static final long serialVersionUID = 0;
private static int consumeTaskSleepTime = 2000;
// 保存任務所須要的數據
private Object threadPoolTaskData;
ThreadPoolTask(Object tasks)
{
this.threadPoolTaskData = tasks;
}
public void run()
{
// 處理一個任務，這裏的處理方式太簡單了，僅僅是一個打印語句
System.out.println(Thread.currentThread().getName());
System.out.println("start .." + threadPoolTaskData);
try
{
// //便於觀察，等待一段時間
Thread.sleep(consumeTaskSleepTime);
}
catch (Exception e)
{
e.printStackTrace();
}
threadPoolTaskData = null;
}
public Object getTask()
{
return this.threadPoolTaskData;
}
}

說明：
一、在這段程序中，一個任務就是一個Runnable類型的對象，也就是一個ThreadPoolTask類型的對象。
二、通常來講任務除了處理方式外，還須要處理的數據，處理的數據經過構造方法傳給任務。
三、在這段程序中，main()方法至關於一個殘忍的領導，他派發出許多任務，丟給一個叫 threadPool的不辭辛苦的小組來作。
這個小組裏面隊員至少有兩個，若是他們兩個忙不過來，任務就被放到任務列表裏面。
若是積壓的任務過多，多到任務列表都裝不下(超過3個)的時候，就僱傭新的隊員來幫忙。可是基於成本的考慮，不能僱傭太多的隊員，至多隻能僱傭 4個。
若是四個隊員都在忙時，再有新的任務，這個小組就處理不了了，任務就會被經過一種策略來處理，咱們的處理方式是不停的派發，直到接受這個任務爲止(更殘忍！呵呵)。
由於隊員工做是須要成本的，若是工做很閒，閒到 3SECONDS都沒有新的任務了，那麼有的隊員就會被解僱了，可是，爲了小組的正常運轉，即便工做再閒，小組的隊員也不能少於兩個。
四、經過調整 produceTaskSleepTime和 consumeTaskSleepTime的大小來實現對派發任務和處理任務的速度的控制，改變這兩個值就能夠觀察不一樣速率下程序的工做狀況。
五、經過調整4中所指的數據，再加上調整任務丟棄策略，換上其餘三種策略，就能夠看出不一樣策略下的不一樣處理方式。
六、對於其餘的使用方法，參看jdk的幫助，很容易理解和使用。

另外一個例子:

點擊(此處)摺疊或打開 this

package demo;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class ThreadPoolExecutorTest
{
private static int queueDeep = 4;
public void createThreadPool()
{
/*
* 建立線程池，最小線程數爲2，最大線程數爲4，線程池維護線程的空閒時間爲3秒，
* 使用隊列深度爲4的有界隊列，若是執行程序還沒有關閉，則位於工做隊列頭部的任務將被刪除，
* 而後重試執行程序（若是再次失敗，則重複此過程），裏面已經根據隊列深度對任務加載進行了控制。
*/
ThreadPoolExecutor tpe = new ThreadPoolExecutor(2, 4, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(queueDeep),
new ThreadPoolExecutor.DiscardOldestPolicy());
// 向線程池中添加 10 個任務
for (int i = 0; i < 10; i++)
{
try
{
Thread.sleep(1);
}
catch (InterruptedException e)
{
e.printStackTrace();
}
while (getQueueSize(tpe.getQueue()) >= queueDeep)
{
System.out.println("隊列已滿，等3秒再添加任務");
try
{
Thread.sleep(3000);
}
catch (InterruptedException e)
{
e.printStackTrace();
}
}
TaskThreadPool ttp = new TaskThreadPool(i);
System.out.println("put i:" + i);
tpe.execute(ttp);
}
tpe.shutdown();
}
private synchronized int getQueueSize(Queue queue)
{
return queue.size();
}
public static void main(String[] args)
{
ThreadPoolExecutorTest test = new ThreadPoolExecutorTest();
test.createThreadPool();
}
class TaskThreadPool implements Runnable
{
private int index;
public TaskThreadPool(int index)
{
this.index = index;
}
public void run()
{
System.out.println(Thread.currentThread() + " index:" + index);
try
{
Thread.sleep(3000);
}
catch (InterruptedException e)
{
e.printStackTrace();
}
}
}
}

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。