Java 高并發(fā)六:JDK并發(fā)包2詳解
1. 線程池的基本使用
1.1.為什么需要線程池
平時的業(yè)務中,如果要使用多線程,那么我們會在業(yè)務開始前創(chuàng)建線程,業(yè)務結束后,銷毀線程。但是對于業(yè)務來說,線程的創(chuàng)建和銷毀是與業(yè)務本身無關的,只關心線程所執(zhí)行的任務。因此希望把盡可能多的cpu用在執(zhí)行任務上面,而不是用在與業(yè)務無關的線程創(chuàng)建和銷毀上面。而線程池則解決了這個問題,線程池的作用就是將線程進行復用。
1.2.JDK為我們提供了哪些支持

JDK中的相關類圖如上圖所示。
其中要提到的幾個特別的類。
Callable類和Runable類相似,但是區(qū)別在于Callable有返回值。
ThreadPoolExecutor是線程池的一個重要實現(xiàn)。
而Executors是一個工廠類。
1.3.線程池的使用
1.3.1.線程池的種類
- new FixedThreadPool 固定數(shù)量的線程池,線程池中的線程數(shù)量是固定的,不會改變。
- new SingleThreadExecutor 單一線程池,線程池中只有一個線程。
- new CachedThreadPool 緩存線程池,線程池中的線程數(shù)量不固定,會根據(jù)需求的大小進行改變。
- new ScheduledThreadPool 計劃任務調度的線程池,用于執(zhí)行計劃任務,比如每隔5分鐘怎么樣,
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
從方法上來看,顯然 FixedThreadPool,SingleThreadExecutor,CachedThreadPool都是ThreadPoolExecutor的不同實例,只是參數(shù)不同。
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}
下面來簡述下 ThreadPoolExecutor構造函數(shù)中參數(shù)的含義。
- corePoolSize 線程池中核心線程數(shù)的數(shù)目
- maximumPoolSize 線程池中最多能容納多少個線程
- keepAliveTime 當現(xiàn)在線程數(shù)目大于corePoolSize時,超過keepAliveTime時間后,多出corePoolSize的那些線程將被終結。
- unit keepAliveTime的單位
- workQueue 當任務數(shù)量很大,線程池中線程無法滿足時,提交的任務會被放到阻塞隊列中,線程空閑下來則會不斷從阻塞隊列中取數(shù)據(jù)。
這樣在來看上面所說的FixedThreadPool,它的線程的核心數(shù)目和最大容納數(shù)目都是一樣的,以至于在工作期間,并不會創(chuàng)建和銷毀線程。當任務數(shù)量很大,線程池中的線程無法滿足時,任務將被保存到LinkedBlockingQueue中,而LinkedBlockingQueue的大小是Integer.MAX_VALUE。這就意味著,任務不斷地添加,會使內(nèi)存消耗越來越大。
而CachedThreadPool則不同,它的核心線程數(shù)量是0,最大容納數(shù)目是Integer.MAX_VALUE,它的阻塞隊列是SynchronousQueue,這是一個特別的隊列,它的大小是0。由于核心線程數(shù)量是0,所以必然要將任務添加到SynchronousQueue中,這個隊列只有一個線程在從中添加數(shù)據(jù),同時另一個線程在從中獲取數(shù)據(jù)時,才能成功。獨自往這個隊列中添加數(shù)據(jù)會返回失敗。當返回失敗時,則線程池開始擴展線程,這就是為什么CachedThreadPool的線程數(shù)目是不固定的。當60s該線程仍未被使用時,線程則被銷毀。
1.4.線程池使用的小例子
1.4.1.簡單線程池
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ThreadPoolDemo {
public static class MyTask implements Runnable {
@Override
public void run() {
System.out.println(System.currentTimeMillis() + "Thread ID:"
+ Thread.currentThread().getId());
try {
Thread.sleep(1000);
} catch (Exception e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) {
MyTask myTask = new MyTask();
ExecutorService es = Executors.newFixedThreadPool(5);
for (int i = 0; i < 10; i++) {
es.submit(myTask);
}
}
}
由于使用的newFixedThreadPool(5),但是啟動了10個線程,所以每次執(zhí)行5個,并且 可以很明顯的看到線程的復用,ThreadId是重復的,也就是前5個任務和后5個任務都是同一批線程去執(zhí)行的。
這里用的是
es.submit(myTask);
還有一種提交方式:
es.execute(myTask);
區(qū)別在于submit會返回一個Future對象,這個將在以后介紹。
1.4.2.ScheduledThreadPool
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class ThreadPoolDemo {
public static void main(String[] args) {
ScheduledExecutorService ses = Executors.newScheduledThreadPool(10);
//如果前面的任務還未完成,則調度不會啟動。
ses.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
try {
Thread.sleep(1000);
System.out.println(System.currentTimeMillis()/1000);
} catch (Exception e) {
// TODO: handle exception
}
}
}, 0, 2, TimeUnit.SECONDS);//啟動0秒后執(zhí)行,然后周期2秒執(zhí)行一次
}
}
輸出:
1454832514
1454832517
1454832520
1454832523
1454832526
...
由于任務執(zhí)行需要1秒,任務調度必須等待前一個任務完成。也就是這里的每隔2秒的意思是,前一個任務完成后2秒再開啟新的一個任務。
2. 擴展和增強線程池
2.1.回調接口
線程池中有一些回調的api來給我們提供擴展的操作。
ExecutorService es = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>()){
@Override
protected void beforeExecute(Thread t, Runnable r) {
System.out.println("準備執(zhí)行");
}
@Override
protected void afterExecute(Runnable r, Throwable t) {
System.out.println("執(zhí)行完成");
}
@Override
protected void terminated() {
System.out.println("線程池退出");
}
};
我們可以通過實現(xiàn)ThreadPoolExecutor的子類去覆蓋ThreadPoolExecutor的beforeExecute,afterExecute,terminated方法來實現(xiàn)在線程執(zhí)行前后,線程池退出時的日志管理或其他操作。
2.2.拒絕策略
有時候,任務非常繁重,導致系統(tǒng)負載太大。在上面說過,當任務量越來越大時,任務都將放到FixedThreadPool的阻塞隊列中,導致內(nèi)存消耗太大,最終導致內(nèi)存溢出。這樣的情況是應該要避免的。因此當我們發(fā)現(xiàn)線程數(shù)量要超過最大線程數(shù)量時,我們應該放棄一些任務。丟棄時,我們應該把任務記下來,而不是直接丟掉。
ThreadPoolExecutor中還有另一個構造函數(shù)。
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;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
threadFactory我們在后面再介紹。
而handler就是拒絕策略的實現(xiàn),它會告訴我們,如果任務不能執(zhí)行了,該怎么做。

共有以上4種策略。
AbortPolicy:如果不能接受任務了,則拋出異常。
CallerRunsPolicy:如果不能接受任務了,則讓調用的線程去完成。
DiscardOldestPolicy:如果不能接受任務了,則丟棄最老的一個任務,由一個隊列來維護。
DiscardPolicy:如果不能接受任務了,則丟棄任務。
ExecutorService es = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(),
new RejectedExecutionHandler() {
@Override
public void rejectedExecution(Runnable r,
ThreadPoolExecutor executor) {
System.out.println(r.toString() + "is discard");
}
});
當然我們也可以自己實現(xiàn)RejectedExecutionHandler接口來自己定義拒絕策略。
2.3.自定義ThreadFactory
剛剛已經(jīng)看到了,在ThreadPoolExecutor的構造函數(shù)中可以指定threadFactory。
線程池中的線程都是由線程工廠創(chuàng)建出來,我們可以自定義線程工廠。
默認的線程工廠:
static class DefaultThreadFactory implements ThreadFactory {
private static final AtomicInteger poolNumber = new AtomicInteger(1);
private final ThreadGroup group;
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final String namePrefix;
DefaultThreadFactory() {
SecurityManager s = System.getSecurityManager();
group = (s != null) ? s.getThreadGroup() :
Thread.currentThread().getThreadGroup();
namePrefix = "pool-" +
poolNumber.getAndIncrement() +
"-thread-";
}
public Thread newThread(Runnable r) {
Thread t = new Thread(group, r,
namePrefix + threadNumber.getAndIncrement(),
0);
if (t.isDaemon())
t.setDaemon(false);
if (t.getPriority() != Thread.NORM_PRIORITY)
t.setPriority(Thread.NORM_PRIORITY);
return t;
}
}
3. ForkJoin
3.1.思想

就是分而治之的思想。
fork/join類似MapReduce算法,兩者區(qū)別是:Fork/Join 只有在必要時如任務非常大的情況下才分割成一個個小任務,而 MapReduce總是在開始執(zhí)行第一步進行分割。看來,F(xiàn)ork/Join更適合一個JVM內(nèi)線程級別,而MapReduce適合分布式系統(tǒng)。
4.2.使用接口
RecursiveAction:無返回值
RecursiveTask:有返回值
4.3.簡單例子
import java.util.ArrayList;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;
public class CountTask extends RecursiveTask<Long>{
private static final int THRESHOLD = 10000;
private long start;
private long end;
public CountTask(long start, long end) {
super();
this.start = start;
this.end = end;
}
@Override
protected Long compute() {
long sum = 0;
boolean canCompute = (end - start) < THRESHOLD;
if(canCompute)
{
for (long i = start; i <= end; i++) {
sum = sum + i;
}
}else
{
//分成100個小任務
long step = (start + end)/100;
ArrayList<CountTask> subTasks = new ArrayList<CountTask>();
long pos = start;
for (int i = 0; i < 100; i++) {
long lastOne = pos + step;
if(lastOne > end )
{
lastOne = end;
}
CountTask subTask = new CountTask(pos, lastOne);
pos += step + 1;
subTasks.add(subTask);
subTask.fork();//把子任務推向線程池
}
for (CountTask t : subTasks) {
sum += t.join();//等待所有子任務結束
}
}
return sum;
}
public static void main(String[] args) {
ForkJoinPool forkJoinPool = new ForkJoinPool();
CountTask task = new CountTask(0, 200000L);
ForkJoinTask<Long> result = forkJoinPool.submit(task);
try {
long res = result.get();
System.out.println("sum = " + res);
} catch (Exception e) {
// TODO: handle exception
e.printStackTrace();
}
}
}
上述例子描述了一個累加和的任務。將累加任務分成100個任務,每個任務只執(zhí)行一段數(shù)字的累加和,最后join后,把每個任務計算出的和再累加起來。
4.4.實現(xiàn)要素
4.4.1.WorkQueue與ctl
每一個線程都會有一個工作隊列
static final class WorkQueue
在工作隊列中,會有一系列對線程進行管理的字段
volatile int eventCount; // encoded inactivation count; < 0 if inactive
int nextWait; // encoded record of next event waiter
int nsteals; // number of steals
int hint; // steal index hint
short poolIndex; // index of this queue in pool
final short mode; // 0: lifo, > 0: fifo, < 0: shared
volatile int qlock; // 1: locked, -1: terminate; else 0
volatile int base; // index of next slot for poll
int top; // index of next slot for push
ForkJoinTask<?>[] array; // the elements (initially unallocated)
final ForkJoinPool pool; // the containing pool (may be null)
final ForkJoinWorkerThread owner; // owning thread or null if shared
volatile Thread parker; // == owner during call to park; else null
volatile ForkJoinTask<?> currentJoin; // task being joined in awaitJoin
ForkJoinTask<?> currentSteal; // current non-local task being executed
這里要注意的是,JDK7和JDK8在ForkJoin的實現(xiàn)上有了很大的差別。我們這里介紹的是JDK8中的。 在線程池中,有時不是所有的線程都在執(zhí)行的,部分線程會被掛起,那些掛起的線程會被存放到一個棧中。內(nèi)部通過一個鏈表表示。
nextWait會指向下一個等待的線程。
poolIndex線程在線程池中的下標索引。
eventCount 在初始化時,eventCount與poolIndex有關??偣?2位,第一位表示是否被激活,15位表示被掛起的次數(shù)
eventCount,剩下的表示poolIndex。用一個字段來表示多個意思。
工作隊列WorkQueue用ForkJoinTask<?>[] array來表示。而top,base來表示隊列的兩端,數(shù)據(jù)在這兩者之間。
在ForkJoinPool中維護著ctl(64位long型)
volatile long ctl;
* Field ctl is a long packed with:
* AC: Number of active running workers minus target parallelism (16 bits)
* TC: Number of total workers minus target parallelism (16 bits)
* ST: true if pool is terminating (1 bit)
* EC: the wait count of top waiting thread (15 bits)
* ID: poolIndex of top of Treiber stack of waiters (16 bits)
AC表示活躍的線程數(shù)減去并行度(大概就是CPU個數(shù))
TC表示總的線程數(shù)減去并行度
ST表示線程池本身是否是激活的
EC表示頂端等待線程的掛起數(shù)
ID表示頂端等待線程的poolIndex
很明顯ST+EC+ID就是我們剛剛所說的 eventCount 。
那么為什么明明5個變量,非要合成一個變量呢。其實用5個變量占用容量也差不多。
用一個變量代碼的可讀性上會差很多。
那么為什么用一個變量呢?其實這點才是最巧妙的地方,因為這5個變量是一個整體,在多線程中,如果用5個變量,那么當修改其中一個變量時,如何保證5個變量的整體性。那么用一個變量則就解決了這個問題。如果用鎖解決,則會降低性能。
用一個變量則保證了數(shù)據(jù)的一致性和原子性。
在ForkJoin中隊ctl的更改都是使用CAS操作,在前面系列的文章中已經(jīng)介紹過,CAS是無鎖的操作,性能很好。
由于CAS操作也只能針對一個變量,所以這種設計是最優(yōu)的。
4.4.2.工作竊取
接下來要介紹下整個線程池的工作流程。
每個線程都會調用runWorker
final void runWorker(WorkQueue w) {
w.growArray(); // allocate queue
for (int r = w.hint; scan(w, r) == 0; ) {
r ^= r << 13; r ^= r >>> 17; r ^= r << 5; // xorshift
}
}
scan()函數(shù)是掃描是否有任務要做。
r是一個相對隨機的數(shù)字。
private final int scan(WorkQueue w, int r) {
WorkQueue[] ws; int m;
long c = ctl; // for consistency check
if ((ws = workQueues) != null && (m = ws.length - 1) >= 0 && w != null) {
for (int j = m + m + 1, ec = w.eventCount;;) {
WorkQueue q; int b, e; ForkJoinTask<?>[] a; ForkJoinTask<?> t;
if ((q = ws[(r - j) & m]) != null &&
(b = q.base) - q.top < 0 && (a = q.array) != null) {
long i = (((a.length - 1) & b) << ASHIFT) + ABASE;
if ((t = ((ForkJoinTask<?>)
U.getObjectVolatile(a, i))) != null) {
if (ec < 0)
helpRelease(c, ws, w, q, b);
else if (q.base == b &&
U.compareAndSwapObject(a, i, t, null)) {
U.putOrderedInt(q, QBASE, b + 1);
if ((b + 1) - q.top < 0)
signalWork(ws, q);
w.runTask(t);
}
}
break;
}
else if (--j < 0) {
if ((ec | (e = (int)c)) < 0) // inactive or terminating
return awaitWork(w, c, ec);
else if (ctl == c) { // try to inactivate and enqueue
long nc = (long)ec | ((c - AC_UNIT) & (AC_MASK|TC_MASK));
w.nextWait = e;
w.eventCount = ec | INT_SIGN;
if (!U.compareAndSwapLong(this, CTL, c, nc))
w.eventCount = ec; // back out
}
break;
}
}
}
return 0;
}
我們接下來看看scan方法,scan的一個參數(shù)是WorkQueue,上面已經(jīng)說過,每個線程都會擁有一個WorkQueue,那么多個線程的WorkQueue就會保存在workQueues里面,r是一個隨機數(shù),通過r來找到某一個 WorkQueue,在WorkQueue里面有所要做的任務。
然后通過WorkQueue的base,取得base的偏移量。
b = q.base
..
long i = (((a.length - 1) & b) << ASHIFT) + ABASE;
..
然后通過偏移量得到最后一個的任務,運行這個任務
t = ((ForkJoinTask<?>)U.getObjectVolatile(a, i))
..
w.runTask(t);
..
通過這個大概的分析理解了過程,我們發(fā)現(xiàn),當前線程調用scan方法后,不會執(zhí)行當前的WorkQueue中的任務,而是通過一個隨機數(shù)r,來得到其他 WorkQueue的任務。這就是ForkJoinPool的主要的一個機理。
當前線程不會只著眼于自己的任務,而是優(yōu)先完成其他任務。這樣做來,防止了饑餓現(xiàn)象的發(fā)生。這樣就預防了某些線程因為卡死或者其他原因而無法及時完成任務,或者某個線程的任務量很大,其他線程卻沒事可做。
然后來看看runTask方法
final void runTask(ForkJoinTask<?> task) {
if ((currentSteal = task) != null) {
ForkJoinWorkerThread thread;
task.doExec();
ForkJoinTask<?>[] a = array;
int md = mode;
++nsteals;
currentSteal = null;
if (md != 0)
pollAndExecAll();
else if (a != null) {
int s, m = a.length - 1;
ForkJoinTask<?> t;
while ((s = top - 1) - base >= 0 &&
(t = (ForkJoinTask<?>)U.getAndSetObject
(a, ((m & s) << ASHIFT) + ABASE, null)) != null) {
top = s;
t.doExec();
}
}
if ((thread = owner) != null) // no need to do in finally clause
thread.afterTopLevelExec();
}
}
有一個有趣的命名:currentSteal,偷得的任務,的確是剛剛解釋的那樣。
task.doExec();
將會完成這個任務。
完成了別人的任務以后,將會完成自己的任務。
通過得到top來獲得自己任務第一個任務
while ((s = top - 1) - base >= 0 && (t = (ForkJoinTask<?>)U.getAndSetObject(a, ((m & s) << ASHIFT) + ABASE, null)) != null)
{
top = s;
t.doExec();
}
接下來,通過一個圖來總結下剛剛線程池的流程

比如有T1,T2兩個線程,T1會通過T2的base來獲得T2的最后一個任務(當然實際上是通過一個隨機數(shù)r來取得某個線程最后一個任務),T1也會通過自己的top來執(zhí)行自己的第一個任務。反之,T2也會如此。
拿其他線程的任務都是從base開始拿的,自己拿自己的任務是從top開始拿的。這樣可以減少沖突
如果沒有找到其他任務
else if (--j < 0) {
if ((ec | (e = (int)c)) < 0) // inactive or terminating
return awaitWork(w, c, ec);
else if (ctl == c) { // try to inactivate and enqueue
long nc = (long)ec | ((c - AC_UNIT) & (AC_MASK|TC_MASK));
w.nextWait = e;
w.eventCount = ec | INT_SIGN;
if (!U.compareAndSwapLong(this, CTL, c, nc))
w.eventCount = ec; // back out
}
break;
}
那么首先會通過一系列運行來改變ctl的值,獲得了nc,然后用CAS將新的值賦值。然后就調用awaitWork()將線程進入等待狀態(tài)(調用的 前面系列文章中提到的unsafe的park方法)。
這里要說明的是改變ctl值這里,首先是將ctl中的AC-1,AC是占ctl的前16位,所以不能直接-1,而是通過AC_UNIT(0x1000000000000)來達到使ctl的前16位-1的效果。
前面說過eventCount中有保存poolIndex,通過poolIndex以及WorkQueue中的nextWait,就能遍歷所有的等待線程。
相關文章
Springboot中的異步任務執(zhí)行及監(jiān)控詳解
這篇文章主要介紹了Springboot中的異步任務執(zhí)行及監(jiān)控詳解,除了自己實現(xiàn)線程外,springboot本身就提供了通過注解的方式,進行異步任務的執(zhí)行,下面主要記錄一下,在Springboot項目中實現(xiàn)異步任務,以及對異步任務進行封裝監(jiān)控,需要的朋友可以參考下2023-10-10
Mybatis配置映射文件中parameterType的用法講解
這篇文章主要介紹了Mybatis配置映射文件中parameterType的用法,具有很好的參考價值,希望對大家有所幫助。如有錯誤或未考慮完全的地方,望不吝賜教2021-09-09
JavaWeb實戰(zhàn)之編寫單元測試類測試數(shù)據(jù)庫操作
這篇文章主要介紹了JavaWeb實戰(zhàn)之編寫單元測試類測試數(shù)據(jù)庫操作,文中有非常詳細的代碼示例,對正在學習javaweb的小伙伴們有很大的幫助,需要的朋友可以參考下2021-04-04
springMVC在restful風格的性能優(yōu)化方案
這篇文章主要介紹了springMVC在restful風格的性能優(yōu)化方案,具有很好的參考價值,希望對大家有所幫助。如有錯誤或未考慮完全的地方,望不吝賜教2021-08-08

