link-JAVA多線程與高并發系列[前言,大綱,目錄]
目錄
-
- 前置知識
-
- Executor:
- ExecutorService:
- Future&Callable
- FutureTask(比較常用)
- CompletableFuture(非常靈活)
- 認識ThreadPoolExecutor
-
- ThreadPoolExecutor的七個重要參數:
- 測試小例子:
- 調整線程池的大小
- ThreadPoolExecutor源碼解析
-
- 1、常用變量的解釋
- 2、構造方法
- 3、送出執行task的過程
- 4、addworker源碼解析
- 5、線程池worker任務單元
- 6、核心線程執行邏輯-runworker
前置知識
關鍵類:
Executor <-extends- ExecutorService <-implements-AbstractExecutorService <-extends- ExecutorService
Callable,Future
Executor:
任務的定義和執行分開,隻有一個執行Runnable方法:
void execute(Runnable command);
ExecutorService:
除了繼承Executor可以執行任務的功能,還完善了整個任務執行器(線程池)的生命周期.比如,一個線程池裡面有很多線程,怎麼送出任務,執行完任務之後應該怎麼處理線程,怎麼關閉等等.
Future&Callable
可以看到ExecutorService裡面有個方法,送出異步的任務:
Future submit(Callable task);
Callable:和Runnable類似,是一個任務,隻不過它執行完後有傳回值,有了傳回值就可以有各種玩法了.
Future:Callable執行完後有一個傳回值,通過Future可以拿到這個結果.
存儲了一個會在将來産生的結果.
看個簡單的例子:
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService service = Executors.newCachedThreadPool();
Future<String> future = service.submit(() -> {
// spend some seconds doing something
TimeUnit.SECONDS.sleep(2);
return "Hello World!";
}); //異步
System.out.println("i will get~");
System.out.println(future.get());//阻塞
service.shutdown();
}
FutureTask(比較常用)
FutureTask implements RunnableFuture,RunnableFuture extend Runnable, Future
FutureTask内部維護了一個Callable成員變量
前面用的Callable隻是一個任務,Future隻是一個傳回值,這個FutureTask就是結合了一下,既是任務又是傳回值.(Apple+pen->pineapple! 😃)
線程池WorkStealingPool和ForkJoinPool用到了FutureTask
看一個小例子:
public static void main(String[] args) throws InterruptedException, ExecutionException {
FutureTask<Integer> task = new FutureTask<>(() -> {
TimeUnit.MILLISECONDS.sleep(500);
return 1000;
}); //new Callable () { Integer call();}
// 可以是線程,也可以是線程池
new Thread(task).start();
System.out.println(task.get()); //阻塞
}
CompletableFuture(非常靈活)
CompletableFuture implements Future,CompletionStage
一個典型的應用場景:
有很多個子系統,他們各自有自己的資料庫存儲系統,可能是MySQL/Oracle/MongoDB等,現在需要統計他們的名額(比如平均請求響應時間)在一張大屏上展示分析.如果串行去查詢子系統的資料,那這個分析的API就執行太久了,但是我們使用CompletableFuture,多線程異步執行,那時間就大大縮短.
當然這個場景用普通線程執行Callable也是可以搞定的,隻是用CompletableFuture比較友善,相當于JDK已經造好輪子了,我們可以直接用它.
show my code:
public class TestCompletableFuture {
public static void main(String[] args) throws ExecutionException, InterruptedException {
normalTest();
futureTest();
// test002();
}
private static void normalTest() {
long start = System.currentTimeMillis();
Map<String, Double> metrics = new HashMap<>(4);
metrics.put("metricsOfMySQL", metricsOfMySQL());
metrics.put("metricsOfOracle", metricsOfOracle());
metrics.put("metricsOfMongoDB", metricsOfMongoDB());
long end = System.currentTimeMillis();
System.out.println("use serial method call! " + (end - start));
System.out.println(metrics + "\n-------------------------------------");
}
private static void futureTest() {
long start = System.currentTimeMillis();
Map<String, Double> metrics = new HashMap<>(4);
CompletableFuture<Double> metricsOfMySQL = CompletableFuture.supplyAsync(TestCompletableFuture::metricsOfMySQL)
.thenApply(value -> metrics.put("metricsOfMySQL", value));
CompletableFuture<Double> metricsOfOracle = CompletableFuture.supplyAsync(TestCompletableFuture::metricsOfOracle)
.thenApply(value -> metrics.put("metricsOfOracle", value));
CompletableFuture<Double> metricsOfMongoDB = CompletableFuture.supplyAsync(TestCompletableFuture::metricsOfMongoDB)
.thenApply(value -> metrics.put("metricsOfMongoDB", value));
System.out.println(metrics);
CompletableFuture.allOf(metricsOfMySQL, metricsOfOracle, metricsOfMongoDB).join();
long end = System.currentTimeMillis();
System.out.println("use completable future! " + (end - start));
System.out.println(metrics);
}
// 其他的用法,靈活的處理結果,有點函數式程式設計的感覺
private static void test002() {
// 異步執行
CompletableFuture.supplyAsync(() -> metricsOfMongoDB())
.thenApply(String::valueOf)
.thenApply(str -> "price " + str)
.thenAccept(System.out::println);
// 阻塞住主線程,等待上面執行完
try {
System.in.read();
} catch (IOException e) {
e.printStackTrace();
}
}
private static double metricsOfMySQL() {
sleepRandom();
return 2.00;
}
private static double metricsOfOracle() {
sleepRandom();
return 3.00;
}
private static double metricsOfMongoDB() {
sleepRandom();
return 1.00;
}
private static void sleepRandom() {
int time = new Random().nextInt(500);
try {
TimeUnit.MILLISECONDS.sleep(time);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.printf("After %s sleep!\n", time);
}
}
}
認識ThreadPoolExecutor
最開始我們建立一個線程,執行一個Runnable任務,執行完後就銷毀線程了.
然而建立一個線程需要跟作業系統申請資源,這個過程是比較耗時的;是以我們最好是讓線程複用,即讓一個線程去持續執行不同的任務,而不是執行一個任務後就銷毀.(可以類比一下泡面桶和陶瓷碗)
線程池裡面不僅僅是線程,它維護這兩個集合,一個是線程集合,一個是任務集合.
ThreadPoolExecutor的七個重要參數:
-
int corePoolSize
核心線程數,最開始線程池裡面沒有線程,來了任務後會先建立一定數量的核心線程去執行任務;一般沒有任務執行時也不會回收核心線程.
-
int maximumPoolSize
最大線程數.當任務比較多,核心線程執行不過來時會放入任務隊列,任務隊列滿了後會建立非核心線程,maximumPoolSize=臨時線程數+核心線程數,主要負責控制臨時線程數.非核心線程在空閑一段時間後會被回收.
-
long keepAliveTime
生存時間.當一個非核心線程很長時間不執行任務了,就銷毀該線程,這個參數就是控制空閑門檻值.
核心線程預設不受此控制,也可以設定參數指定核心線程受此控制(allowCoreThreadTimeOut).
-
TimeUnit unit
生存時間機關,見名知意
-
BlockingQueue workQueue
任務隊列
-
ThreadFactory threadFactory
線程工廠,自定義建立線程的方式.
有個預設的DefaultThreadFactory,指定了線程名字,daemon=false,priority=5(NORM_PRIORITY).不要小看線程名,多線程環境追蹤錯誤日志時大有用處.
-
RejectedExecutionHandler handler
拒絕政策,當任務很多,任務隊列滿了,非核心線程數也達到上限後,再來任務的時候的處理政策.拒絕政策可以自定義,JDK提供了四種拒絕政策:
- AbortPolicy:抛異常,這也是預設的拒絕政策
- DiscardPolicy:安靜的丢掉
-
DiscardOldestPolicy:丢掉隊列中最老的任務,把新的放入隊列
做遊戲的時候可能會用,比如一個角色的每次移動作為一個操作當如線程池中,正常情況是依次移動;當隊列滿了就把最老的丢掉,減少影響.
-
CallerRunsPolicy:送出任務者(調用execute的線程)處理該任務
(實戰中這四種一般都不用,而是自定義)
阿裡開發手冊1.5.0裡面一丶(六)也講到,很多關于線程的規範,下面列舉幾條:
- 線程資源必須通過線程池提供，不允許在應用中自行顯式建立線程。
-
線程池不允許使用 Executors 去建立，而是通過ThreadPoolExecutor 的方式，這樣的處理方式讓寫的同學更加明确線程池的運作規則，規避資源耗盡的風險。
說明：Executors 傳回的線程池對象的弊端如下：
1） FixedThreadPool 和 SingleThreadPool：
允許的請求隊列長度為 Integer.MAX_VALUE，可能會堆積大量的請求,進而導緻 OOM。
2） CachedThreadPool：
允許的建立線程數量為 Integer.MAX_VALUE，可能會建立大量的線程,進而導緻 OOM。
- 建立線程或線程池時請指定有意義的線程名稱，友善出錯時回溯。
測試小例子:
public class T05_00_HelloThreadPool {
static class Task implements Runnable {
private int i;
public Task(int i) {
this.i = i;
}
@Override
public void run() {
// 列印一下目前線程
System.out.println(Thread.currentThread().getName() + " Task " + i);
try {
// 阻塞住,以便認識不同的拒絕政策
TimeUnit.DAYS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Override
public String toString() {
return "Task{" +
"i=" + i +
'}';
}
}
public static void main(String[] args) {
// 初始化一個線程池,最多同時接納8個任務
ThreadPoolExecutor tpe = new ThreadPoolExecutor(2, 4,
60, TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(4),
Executors.defaultThreadFactory(),
new ThreadPoolExecutor.CallerRunsPolicy()
);
// 把線程池占滿
for (int i = 0; i < 8; i++) {
tpe.execute(new Task(i));
}
// 列印一下目前線程等待隊列,是線程2,3,4,5;
// 因為0和1正被核心線程執行,6和7被非核心線程執行
System.out.println(tpe.getQueue());
tpe.execute(new Task(100));
// 如果是DiscardOldestPolicy,會發現任務2被丢掉了,任務100加入等待隊列
// 如果是CallerRunsPolicy,這句話不會列印,因為新的任務被主線程執行,而任務會阻塞線程;但是會列印main Task 100
System.out.println("main thread end\n" + tpe.getQueue());
// 不再接收新任務,等已有任務執行完後關掉線程池
tpe.shutdown();
// 嘗試馬上關掉線程池,不等目前任務結束,而是通過Thread.interrupt打斷線程
// 如果線程沒有正确處理InterruptedException,那就永遠那不會被終結
// tpe.shutdownNow();
}
}
調整線程池的大小
下面是一個建議,也可以說是标準公式吧,但是這個公式中的等待時間和預估時間的比率很難預估出來,工程中還是需要經過各種情況的壓力測試,然後取一個相對各方面都照顧的到的值.
一般的等待時間都花在IO上,是以W/C比較高時也稱為IO密集型.
ThreadPoolExecutor源碼解析
這塊扣起來賊頭疼,我們先領會戰略精神,具體戰術日後再議…
昨天看AQS的源碼,扣了半天沒搞明白,浪費好多時間,還有很多"上天入地"的任務待完成…
這裡補充記錄一點,JAVA(不知道其他語言怎麼說…)中整數的表示形式,以4位的數來說:
1.正整數和0,就是正常的二進制,1就是0001,2就是0010
2.負整數=對應正整數的反碼的補碼,-1反碼->1110補碼->1111,即十進制-1的二進制為1111
1、常用變量的解釋
// 1. `ctl`，可以看做一個int類型的數字，高3位表示線程池狀态，低29位表示worker數量
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 2. `COUNT_BITS`，`Integer.SIZE`為32，是以`COUNT_BITS`為29
private static final int COUNT_BITS = Integer.SIZE - 3;
// 3. `CAPACITY`，線程池允許的最大線程數。1左移29位，然後減1，即為 2^29 - 1
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// runState is stored in the high-order bits
// 4. 線程池有5種狀态，按大小排序如下：RUNNING < SHUTDOWN < STOP < TIDYING < TERMINATED
private static final int RUNNING = -1 << COUNT_BITS;
private static final int SHUTDOWN = 0 << COUNT_BITS;
private static final int STOP = 1 << COUNT_BITS;
private static final int TIDYING = 2 << COUNT_BITS;
private static final int TERMINATED = 3 << COUNT_BITS;
// Packing and unpacking ctl
// 5. `runStateOf()`，擷取線程池狀态，通過按位與操作，低29位将全部變成0
private static int runStateOf(int c) { return c & ~CAPACITY; }
// 6. `workerCountOf()`，擷取線程池worker數量，通過按位與操作，高3位将全部變成0
private static int workerCountOf(int c) { return c & CAPACITY; }
// 7. `ctlOf()`，根據線程池狀态和線程池worker數量，生成ctl值
private static int ctlOf(int rs, int wc) { return rs | wc; }
/*
* Bit field accessors that don't require unpacking ctl.
* These depend on the bit layout and on workerCount being never negative.
*/
// 8. `runStateLessThan()`，線程池狀态小于xx
private static boolean runStateLessThan(int c, int s) {
return c < s;
}
// 9. `runStateAtLeast()`，線程池狀态大于等于xx
private static boolean runStateAtLeast(int c, int s) {
return c >= s;
}
2、構造方法
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
// 基本類型參數校驗
if (corePoolSize < 0 ||
maximumPoolSize <= 0 ||
maximumPoolSize < corePoolSize ||
keepAliveTime < 0)
throw new IllegalArgumentException();
// 空指針校驗
if (workQueue == null || threadFactory == null || handler == null)
throw new NullPointerException();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
// 根據傳入參數`unit`和`keepAliveTime`，将存活時間轉換為納秒存到變量`keepAliveTime `中
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
3、送出執行task的過程
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
/*
* Proceed in 3 steps:
*
* 1. If fewer than corePoolSize threads are running, try to
* start a new thread with the given command as its first
* task. The call to addWorker atomically checks runState and
* workerCount, and so prevents false alarms that would add
* threads when it shouldn't, by returning false.
*
* 2. If a task can be successfully queued, then we still need
* to double-check whether we should have added a thread
* (because existing ones died since last checking) or that
* the pool shut down since entry into this method. So we
* recheck state and if necessary roll back the enqueuing if
* stopped, or start a new thread if there are none.
*
* 3. If we cannot queue task, then we try to add a new
* thread. If it fails, we know we are shut down or saturated
* and so reject the task.
*/
int c = ctl.get();
// worker數量比核心線程數小，直接建立worker執行任務
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
// worker數量超過核心線程數，任務直接進入隊列
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
// 線程池狀态不是RUNNING狀态，說明執行過shutdown指令，需要對新加入的任務執行reject()操作。
// 這兒為什麼需要recheck，是因為任務入隊列前後，線程池的狀态可能會發生變化。
if (! isRunning(recheck) && remove(command))
reject(command);
// 這兒為什麼需要判斷0值，主要是線上程池構造方法中，核心線程數允許為0
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
// 如果線程池不是運作狀态，或者任務進入隊列失敗，則嘗試建立worker執行任務。
// 這兒有3點需要注意：
// 1. 線程池不是運作狀态時，addWorker内部會判斷線程池狀态
// 2. addWorker第2個參數表示是否建立核心線程
// 3. addWorker傳回false，則說明任務執行失敗，需要執行reject操作
else if (!addWorker(command, false))
reject(command);
}
4、addworker源碼解析
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
// 外層自旋
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 這個條件寫得比較難懂，我對其進行了調整，和下面的條件等價
// (rs > SHUTDOWN) ||
// (rs == SHUTDOWN && firstTask != null) ||
// (rs == SHUTDOWN && workQueue.isEmpty())
// 1. 線程池狀态大于SHUTDOWN時，直接傳回false
// 2. 線程池狀态等于SHUTDOWN，且firstTask不為null，直接傳回false
// 3. 線程池狀态等于SHUTDOWN，且隊列為空，直接傳回false
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
// 内層自旋
for (;;) {
int wc = workerCountOf(c);
// worker數量超過容量，直接傳回false
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// 使用CAS的方式增加worker數量。
// 若增加成功，則直接跳出外層循環進入到第二部分
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
// 線程池狀态發生變化，對外層循環進行自旋
if (runStateOf(c) != rs)
continue retry;
// 其他情況，直接内層循環進行自旋即可
// else CAS failed due to workerCount change; retry inner loop
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
// worker的添加必須是串行的，是以需要加鎖
mainLock.lock();
try {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
// 這兒需要重新檢查線程池狀态
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
// worker已經調用過了start()方法，則不再建立worker
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
// worker建立并添加到workers成功
workers.add(w);
// 更新`largestPoolSize`變量
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
// 啟動worker線程
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
// worker線程啟動失敗，說明線程池狀态發生了變化（關閉操作被執行），需要進行shutdown相關操作
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
5、線程池worker任務單元
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
/**
* This class will never be serialized, but we provide a
* serialVersionUID to suppress a javac warning.
*/
private static final long serialVersionUID = 6138294804551838833L;
/** Thread this worker is running in. Null if factory fails. */
final Thread thread;
/** Initial task to run. Possibly null. */
Runnable firstTask;
/** Per-thread task counter */
volatile long completedTasks;
/**
* Creates with given first task and thread from ThreadFactory.
* @param firstTask the first task (null if none)
*/
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
// 這兒是Worker的關鍵所在，使用了線程工廠建立了一個線程。傳入的參數為目前worker
this.thread = getThreadFactory().newThread(this);
}
/** Delegates main run loop to outer runWorker */
public void run() {
runWorker(this);
}
// 省略代碼...
}
6、核心線程執行邏輯-runworker
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
// 調用unlock()是為了讓外部可以中斷
w.unlock(); // allow interrupts
// 這個變量用于判斷是否進入過自旋（while循環）
boolean completedAbruptly = true;
try {
// 這兒是自旋
// 1. 如果firstTask不為null，則執行firstTask；
// 2. 如果firstTask為null，則調用getTask()從隊列擷取任務。
// 3. 阻塞隊列的特性就是：當隊列為空時，目前線程會被阻塞等待
while (task != null || (task = getTask()) != null) {
// 這兒對worker進行加鎖，是為了達到下面的目的
// 1. 降低鎖範圍，提升性能
// 2. 保證每個worker執行的任務是串行的
w.lock();
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
// 如果線程池正在停止，則對目前線程進行中斷操作
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
// 執行任務，且在執行前後通過`beforeExecute()`和`afterExecute()`來擴充其功能。
// 這兩個方法在目前類裡面為空實作。
try {
beforeExecute(wt, task);
Throwable thrown = null;
try {
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
afterExecute(task, thrown);
}
} finally {
// 幫助gc
task = null;
// 已完成任務數加一
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
// 自旋操作被退出，說明線程池正在結束
processWorkerExit(w, completedAbruptly);
}
}
![Java小案例——随機數猜測随機數猜測[圖]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)