並行程式設計從入門到放棄系列開始和結束
對於 Java 部分的面試來說，突然想到並行這一塊的內容是不太完整的，這篇文章會通篇把多執行緒和並行都大致闡述一遍，至少能夠達到了解原理和使用的目的，內容會比較多，從最基本的執行緒到我們常用的類會統一說一遍，慢慢看。
程序&執行緒
對於基本的概念，大家應該都很熟悉了，程序是資源分配的單位，執行緒是CPU排程的單位，執行緒是程序中的一個實體。
對於我們的Java程式來說，天生就是多執行緒的，我們通過main方法啟動，就是啟動了一個JVM的程序，同時建立一個名為main的執行緒，main就是JVM程序中的一個實體執行緒。
執行緒生命週期
執行緒幾種基本狀態：
- New，初始狀態，就是New了一個執行緒，但是還沒有呼叫start方法
- Runnable，可執行Ready或者執行Running狀態，執行緒的就緒和執行中狀態我們統稱為Runnable執行狀態
- Blocked/Wating/Timed_Wating，這些狀態統一就叫做休眠狀態
- Terminated，終止狀態
幾個狀態之間的轉換我們分別來說。
New：我們建立一個執行緒，但是執行緒沒有呼叫start方法，就是初始化狀態。
Runnable：呼叫start()啟動執行緒進入Ready可執行狀態，等待CPU排程之後進入到Running狀態。
Blocked：阻塞狀態，當執行緒在等待進入synchronized鎖的時候，進入阻塞狀態。
Waiting：等待狀態需要被顯示的喚醒，進入該狀態分為三種情況，在synchonized中呼叫Object.wait()，呼叫Thread.join()，呼叫LockSupport.park()。
Timed_Waiting：和Waiting的區別就是多了超時時間，不需要顯示喚醒，達到超時時間之後自動喚醒，呼叫圖中的一些帶有超時引數的方法則會進入該狀態。
Terminated：終止狀態，執行緒執行完畢。
守護執行緒&使用者執行緒
Java中的執行緒分為守護執行緒和使用者執行緒，上面我們提到的main執行緒其實就是一個使用者執行緒。
他們最主要的區別就在於，只要有非守護執行緒沒有結束，JVM就不會正常退出，而守護執行緒則不會影響JVM的退出。
可以通過簡單的方法設定一個執行緒為守護執行緒。
Thread t = new Thread();
t.setDaemon(true);
鎖
鎖是控制多執行緒並行存取共用資源的方式，為了更簡單快速的瞭解Java中的鎖，我們可以按照顯示鎖和隱式鎖來做一個大致的區分。
隱式鎖
在沒有Lock介面之前，加鎖通過synchronzied實現，在之前的Java基礎系列中我已經說過了，就不在這裡過多的闡述，此處參照之前寫過的，更多詳細可以看《我想進大廠》之Java基礎奪命連環16問。
synchronized是java提供的原子性內建鎖，這種內建的並且使用者看不到的鎖也被稱為監視器鎖，使用synchronized之後，會在編譯之後在同步的程式碼塊前後加上monitorenter和monitorexit位元組碼指令，他依賴作業系統底層互斥鎖實現，主要作用就是實現原子性操作和解決共用變數的記憶體可見性問題。
執行monitorenter指令時會嘗試獲取物件鎖，如果物件沒有被鎖定或者已經獲得了鎖，鎖的計數器+1。此時其他競爭鎖的執行緒則會進入等待佇列中。
執行monitorexit指令時則會把計數器-1，當計數器值為0時，則鎖釋放，處於等待佇列中的執行緒再繼續競爭鎖。
如果再深入到原始碼來說，synchronized實際上有兩個佇列waitSet和entryList。
- 當多個執行緒進入同步程式碼塊時，首先進入entryList
- 有一個執行緒獲取到monitor鎖後，就賦值給當前執行緒，並且計數器+1
- 如果執行緒呼叫wait方法，將釋放鎖，當前執行緒置為null，計數器-1，同時進入waitSet等待被喚醒，呼叫notify或者notifyAll之後又會進入entryList競爭鎖
- 如果執行緒執行完畢，同樣釋放鎖，計數器-1，當前執行緒置為null
顯示鎖
雖然synchronized使用簡單，但是也使得加鎖的流程固化了，顯示鎖在Java1.5版本之後加入了Lock介面，可以通過宣告式顯示的加鎖和解鎖。
Lock lock = new ReentrantLock();
lock.lock(); //加鎖
lock.unlock(); //解鎖
獨佔鎖
在上述的虛擬碼中，我們使用到了ReentrantLock，它其實就是獨佔鎖，獨佔鎖保證任何時候都只有一個執行緒能獲得鎖，當然了，synchronized也是獨佔鎖。
這裡我們看ReentrantLock的幾個加鎖介面。
void lock(); //阻塞加鎖
void lockInterruptibly() throws InterruptedException; //可中斷
boolean tryLock(); //非阻塞
boolean tryLock(long time, TimeUnit unit) throws InterruptedException; //超時加鎖
這幾個加鎖介面，向我們明白地展示了他和synchronized的區別。
- 可中斷加鎖
lockInterruptibly，synchronized可能會有死鎖的問題，那麼解決方案就是能響應中斷。當前執行緒加鎖時，如果其他執行緒呼叫當前執行緒的中斷方法，則會丟擲異常。 - 非阻塞加鎖
tryLock，呼叫後立刻返回，獲取鎖則返回true，否則返回false - 支援超時加鎖
tryLock(long time, TimeUnit unit)，超時時間內獲取鎖返回true，否則返回false - 支援公平和非公平鎖，公平指的是獲取鎖按照請求鎖的時間順序決定，先到先得，非公平則是直接競爭鎖，先到不一定先得
- 支援Condition
如果你看過阻塞佇列的原始碼，那麼你對 Condition 應該挺了解了，我們舉個栗子來看看，我們需要實現：
- 如果佇列滿了，那麼寫入阻塞
- 如果佇列空了，那麼刪除（取元素）阻塞
我們給阻塞佇列提供一個 put 寫入元素和 take 刪除元素的方法。
put 時候加鎖且響應中斷，如果佇列滿了，notFull.await 釋放鎖，進入阻塞狀態，反之，則把元素新增到佇列中，notEmpty.signal 喚醒阻塞在刪除元素的執行緒。
take 的時候一樣加鎖且響應中斷，如果佇列空了，notEmpty.await 進入釋放鎖，進入阻塞狀態，反之，則刪除元素，notFull.signal 喚醒阻塞在新增元素的執行緒。
public class ConditionTest {
public static void main(String[] args) throws Exception {
ArrayBlockingQueue arrayBlockingQueue = new ArrayBlockingQueue(10);
}
static class ArrayBlockingQueue<E> {
private Object[] items;
int takeIndex;
int putIndex;
int count;
private ReentrantLock lock;
private Condition notEmpty;
private Condition notFull;
public ArrayBlockingQueue(int capacity) {
this.items = new Object[capacity];
lock = new ReentrantLock();
notEmpty = lock.newCondition();
notFull = lock.newCondition();
}
public void put(E e) throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == items.length) {
notFull.await();
}
enqueue(e);
} finally {
lock.unlock();
}
}
private void enqueue(E x) {
final Object[] items = this.items;
items[putIndex] = x;
if (++putIndex == items.length){
putIndex = 0;
}
count++;
notEmpty.signal();
}
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0) {
notEmpty.await();
}
return dequeue();
} finally {
lock.unlock();
}
}
private E dequeue() {
final Object[] items = this.items;
E x = (E) items[takeIndex];
items[takeIndex] = null;
if (++takeIndex == items.length){
takeIndex = 0;
}
count--;
notFull.signal();
return x;
}
}
}
讀寫鎖
讀寫鎖，也可以稱作共用鎖，區別於獨佔鎖，共用鎖則可以允許多個執行緒同時持有，如ReentrantReadWriteLock允許多執行緒並行讀，要簡單概括就是：讀讀不互斥，讀寫互斥，寫寫互斥。
ReentrantReadWriteLock
通過閱讀原始碼發現它內部維護了兩個鎖：讀鎖和寫鎖。
private final ReentrantReadWriteLock.ReadLock readerLock;
private final ReentrantReadWriteLock.WriteLock writerLock;
本質上，不管是ReentrantLock還是ReentrantReadWriteLock都是基於AQS，AQS只有一個狀態位state，對於ReentrantReadWriteLock實現讀鎖和寫鎖則是對state做出了區分，高16位元表示的是讀鎖的狀態，低16表示的是寫鎖的狀態。
我們可以看一個原始碼中給出的使用例子。
class CacheData {
Object data;
volatile boolean cacheValid;
final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
void processCachedData() {
rwl.readLock().lock();
if (!cacheValid) {
// 必須先釋放讀鎖，再加寫鎖
rwl.readLock().unlock();
rwl.writeLock().lock();
try {
// 重新校驗狀態，防止並行問題
if (!cacheValid) {
data = ...
cacheValid = true;
}
// 寫鎖降級為讀鎖
rwl.readLock().lock();
} finally {
rwl.writeLock().unlock(); // 寫鎖釋放，仍然持有讀鎖
}
} try {
use(data);
} finally {
rwl.readLock().unlock();
}
}
}
這個例子巢狀寫的其實不太好理解，因為他包含了一個寫鎖降級的概念，實際上我們自己寫最簡單的例子就是這樣，例子中給到的範例其實是一個意思，只是在寫鎖釋放前先降級為讀鎖，明白意思就好。
rwl.readLock().lock();
doSomething();
rwl.readLock().unlock();
rwl.writeLock().lock();
doSomething();
rwl.writeLock().unlock();
額外需要注意的是，寫鎖可以降級為讀鎖，但是讀鎖不能升級為寫鎖，比如下面這種寫法是不支援的。
rwl.readLock().lock();
doSomething();
rwl.writeLock().lock();
doSomething();
rwl.writeLock().unlock();
rwl.readLock().unlock();
StampedLock
這是JDK1.8之後新增的一個鎖，相比ReentrantReadWriteLock他的效能更好，在讀鎖和寫鎖的基礎上增加了一個樂觀讀鎖。
寫鎖：他的寫鎖基本上和ReentrantReadWriteLock一樣，但是不可重入。
讀鎖：也和ReentrantReadWriteLock一樣，但是不可重入。
樂觀讀鎖：普通的讀鎖通過CAS去修改當前state狀態，樂觀鎖實現原理則是加鎖的時候返回一個stamp（鎖狀態），然後還需要呼叫一次validate(stamp)判斷當前是否有其他執行緒持有了寫鎖，通過的話則可以直接運算元據，反之升級到普通的讀鎖，之前我們說到讀寫鎖也是互斥的，那麼樂觀讀和寫就不是這樣的了，他能支援一個執行緒去寫。所以，他效能更高的原因就來自於沒有CAS的操作，只是簡單的位運算拿到當前的鎖狀態stamp，並且能支援另外的一個執行緒去寫。
總結下來可以理解為：讀讀不互斥，讀寫不互斥，寫寫互斥，另外通過tryConvertToReadLock()和tryConvertToWriteLock()等方法支援鎖的升降級。
還是按照官方的檔案舉個栗子，方便理解，兩個方法分別表示樂觀鎖的使用和鎖升級的使用。
public class StampedLockTest {
private double x, y;
private final StampedLock sl = new StampedLock();
double distanceFromOrigin() {
// 樂觀鎖
long stamp = sl.tryOptimisticRead();
double currentX = x, currentY = y;
if (!sl.validate(stamp)) {
//狀態已經改變，升級到讀鎖，重新讀取一次最新的資料
stamp = sl.readLock();
try {
currentX = x;
currentY = y;
} finally {
sl.unlockRead(stamp);
}
}
return Math.sqrt(currentX * currentX + currentY * currentY);
}
void moveIfAtOrigin(double newX, double newY) {
// 可以使用樂觀鎖替代
long stamp = sl.readLock();
try {
while (x == 0.0 && y == 0.0) {
// 嘗試升級到寫鎖
long ws = sl.tryConvertToWriteLock(stamp);
if (ws != 0L) {
//升級成功，替換當前stamp標記
stamp = ws;
x = newX;
y = newY;
break;
} else {
//升級失敗，再次獲取寫鎖
sl.unlockRead(stamp);
stamp = sl.writeLock();
}
}
} finally {
sl.unlock(stamp);
}
}
}
LockSupport
LockSupport是一個比較基礎的工具類，基於Unsafe實現，主要就是提供執行緒阻塞和喚醒的能力，上面我們提到對執行緒生命週期狀態的時候也說過了，LockSupport的幾個park功能將會把執行緒阻塞，直到被喚醒。
看看他的幾個核心方法：
public static void park(); //阻塞當前執行緒
public static void parkNanos(long nanos); //阻塞當前執行緒加上了超時時間，達到超時時間之後返回
public static void parkUntil(long deadline); //和上面類似，引數deadline代表的是從1970到現在時間的毫秒數
public static void unpark(Thread thread);// 喚醒執行緒
舉個栗子：
public class Test {
public static void main(String[] args) throws Exception {
int sleepTime = 3000;
Thread t = new Thread(() -> {
System.out.println(Thread.currentThread().getName() + "掛起");
LockSupport.park();
System.out.println(Thread.currentThread().getName() + "繼續工作");
});
t.start();
System.out.println("主執行緒sleep" + sleepTime);
Thread.sleep(sleepTime);
System.out.println("主執行緒喚醒阻塞執行緒");
LockSupport.unpark(t);
}
}
//輸出如下
主執行緒sleep3000
Thread-0掛起
主執行緒喚醒阻塞執行緒
Thread-0繼續工作
原子類
多執行緒環境下操作變數，除了可以用我們上面一直說的加鎖的方式，還有其他更簡單快捷的辦法嗎？
JDK1.5之後引入的原子操作包下面的一些類提供給了我們一種無鎖操作變數的方式，這種通過CAS操作的方式更高效並且執行緒安全。
基本資料型別
我們先說針對基本資料型別提供的AtomicInteger、AtomicLong、AtomicBoolean，看名字都知道是幹嘛的，由於基本上沒什麼區別，以AtomicInteger的方法舉例來說明。
public final int getAndIncrement(); //舊值+1，返回舊值
public final int getAndDecrement(); //舊值-1，返回舊值
public final int getAndAdd(int delta); //舊值+delta，返回舊值
public final int getAndSet(int newValue); //舊值設定為newValue，返回舊值
public final int getAndAccumulate(int x,IntBinaryOperator accumulatorFunction); //舊值根據傳入方法進行計算，返回舊值
public final int getAndUpdate(IntUnaryOperator updateFunction)//舊值根據傳入進行計算，返回舊值
與之相對應的還有一套方法比如incrementAndGet()等等，規則完全一樣，只是返回的是新值。
我們看看下面的例子，針對自定義規則傳參，比如我們可以把計算規則改成乘法。
public class AtomicIntegerTest {
public static void main(String[] args) {
AtomicInteger atomic = new AtomicInteger(10);
System.out.println(atomic.getAndIncrement()); //10
System.out.println(atomic.getAndDecrement()); //11
System.out.println(atomic.getAndAdd(2));//10
System.out.println(atomic.getAndSet(10)); //12
System.out.println(atomic.get()); //10
System.out.println("=====================");
System.out.println(atomic.getAndAccumulate(3, (left, right) -> left * right)); // 10
System.out.println(atomic.get()); //30
System.out.println(atomic.getAndSet(10)); //30
System.out.println("=====================");
System.out.println(atomic.getAndUpdate(operand -> operand * 20)); // 10
System.out.println(atomic.get()); //200
}
}
另外提到一嘴，基本資料型別只給了Integer、Long、Boolean，那其他的基本資料型別呢？其實看下AtomicBoolean的原始碼我們發現其實他本質上是轉成了Integer處理的，那麼針對其他的型別也可以參考這個思路來實現。
陣列
針對陣列型別的原子操作提供了3個，可以方便的更新陣列中的某個元素。
AtomicIntegerArray：針對Integer陣列的原子操作。
AtomicLongArray：針對Long陣列的原子操作。
AtomicReferenceArray：針對參照型別陣列的原子操作。
和上面說的Atomic其實也沒有太大的區別，還是以AtomicIntegerArray舉例說明，主要方法也基本一樣。
public final int getAndIncrement(int i);
public final int getAndDecrement(int i);
public final int getAndAdd(int i, int delta);
public final int getAndSet(int i, int newValue);
public final int getAndAccumulate(int i, int x,IntBinaryOperator accumulatorFunction);
public final int getAndUpdate(int i, IntUnaryOperator updateFunction);
操作一模一樣，只是多了一個參數列示當前索引的位置，同樣有incrementAndGet等一套方法，返回最新值，沒有區別，對於參照型別AtomicReferenceArray來說只是沒有了increment和decrement這些方法，其他的也都大同小異，不再贅述。
說實話，這個都沒有舉栗子的必要。
public class AtomicIntegerArrayTest {
public static void main(String[] args) {
int[] array = {10};
AtomicIntegerArray atomic = new AtomicIntegerArray(array);
System.out.println(atomic.getAndIncrement(0)); //10
System.out.println(atomic.get(0));//11
System.out.println(atomic.getAndDecrement(0)); //11
System.out.println(atomic.getAndAdd(0, 2));//10
System.out.println(atomic.getAndSet(0, 10)); //12
System.out.println(atomic.get(0)); //10
System.out.println("=====================");
System.out.println(atomic.getAndAccumulate(0, 3, (left, right) -> left * right)); // 10
System.out.println(atomic.get(0)); //30
System.out.println(atomic.getAndSet(0, 10)); //30
System.out.println("=====================");
System.out.println(atomic.getAndUpdate(0, operand -> operand * 20)); // 10
System.out.println(atomic.get(0)); //200
}
}
參照型別
像AtomicInteger那種，只能原子更新一個變數，如果需要同時更新多個變數，就需要使用我們的參照型別的原子類，針對參照型別的原子操作提供了3個。
AtomicReference：針對參照型別的原子操作。
AtomicMarkableReference：針對帶有標記位的參照型別的原子操作。
AtomicStampedReference：針對帶有標記位的參照型別的原子操作。
AtomicMarkableReference和AtomicStampedReference非常類似，他們是為了解決CAS中的ABA的問題（別說你不知道啥是ABA問題），只不過這個標記的型別不同，我們看下原始碼。
AtomicMarkableReference標記型別是布林型別，所以其實他版本就倆，true和false。
AtomicMarkableReference標記型別是整型，那可不就是正常的版本號嘛。
public class AtomicMarkableReference<V> {
private static class Pair<T> {
final T reference;
final boolean mark; //標記
}
}
public class AtomicStampedReference<V> {
private static class Pair<T> {
final T reference;
final int stamp; // 標記
}
}
方法還是那幾個，老樣子。
public final V getAndSet(V newValue);
public final V getAndUpdate(UnaryOperator<V> updateFunction);
public final V getAndAccumulate(V x, BinaryOperator<V> accumulatorFunction);
public final boolean compareAndSet(V expect, V update);
簡單舉個栗子：
public class AtomicReferenceTest {
public static void main(String[] args) {
User user = new User(1L, "test", "test");
AtomicReference<User> atomic = new AtomicReference<>(user);
User pwdUpdateUser = new User(1L,"test","newPwd");
System.out.println(atomic.getAndSet(pwdUpdateUser));
System.out.println(atomic.get());
}
@NoArgsConstructor
@AllArgsConstructor
@Data
@ToString
static class User {
private Long id;
private String username;
private String password;
}
}
//輸出
AtomicReferenceTest.User(id=1, username=test, password=test)
AtomicReferenceTest.User(id=1, username=test, password=newPwd)
物件屬性
針對物件屬性的原子操作也還是提供了3個。
AtomicIntegerFieldUpdater：針對參照型別裡的整型屬性的原子操作。
AtomicLongFieldUpdater：針對參照型別裡的長整型屬性的原子操作。
AtomicReferenceFieldUpdater：針對參照型別裡的屬性的原子操作。
需要注意的是，需要更新的屬性欄位不能是private，並且必須用volatile修飾，否則會報錯。
舉個栗子：
public class AtomicReferenceFieldTest {
public static void main(String[] args) {
AtomicReferenceFieldUpdater<User, String> atomic = AtomicReferenceFieldUpdater.newUpdater(User.class, String.class, "password");
User user = new User(1L, "test", "test");
System.out.println(atomic.getAndSet(user, "newPwd"));
System.out.println(atomic.get(user));
}
@NoArgsConstructor
@AllArgsConstructor
@Data
@ToString
static class User {
private Long id;
private String username;
volatile String password;
}
}
//輸出
test
newPwd
累加器
累加器有4個，都來自JDK1.8新增的，為啥新增呢？因為Doug大佬覺得AtomicLong還不夠快，雖然說通過CAS操作已經很快了，但是眾所知周，高並行同時操作一個共用變數只有一個成功，那其他的執行緒都在無限自旋，大量的浪費了CPU的資源，所以累加器Accumulator的思路就是把一個變數拆成多個變數，這樣多執行緒去操作競爭多個變數資源，效能不就提升了嘛。
也就是說，在高並行的場景下，可以儘量的使用下面這些類來替換基礎型別操作的那些AtomicLong之類的，可以提高效能。
LongAdder：Long型別的累加，LongAccumulator的特例。
LongAccumulator：Long型別的累加。
DoubleAdder：Double型別的累加，DoubleAccumulator的特例。
DoubleAccumulator:Double型別的累加。
由於LongAdder和DoubleAdder都是一樣的，我們以LongAdder和LongAccumulator舉例來說明它的一些簡單的原理。
LongAdder
它繼承自Striped64，內部維護了一個Cell陣列，核心思想就是把單個變數的競爭拆分，多執行緒下如果一個Cell競爭失敗，轉而去其他Cell再次CAS重試。
transient volatile Cell[] cells;
transient volatile long base;
在計算當前值的時候，則是累加所有cell的value再加上base。
public long sum() {
Cell[] as = cells; Cell a;
long sum = base;
if (as != null) {
for (int i = 0; i < as.length; ++i) {
if ((a = as[i]) != null)
sum += a.value;
}
}
return sum;
}
這裡還涉及到一個偽共用的概念，至於啥是偽共用，看看之前我寫的真實位元組二面：什麼是偽共用？。
解決偽共用的真正的核心就在Cell陣列，可以看到，Cell陣列使用了Contented註解。
@sun.misc.Contended static final class Cell {
volatile long value;
Cell(long x) { value = x; }
}
在上面我們提到陣列的記憶體地址都是連續的，所以陣列內的元素經常會被放入一個快取行，這樣的話就會帶來偽共用的問題，影響效能，這裡使用Contented進行填充，就避免了偽共用的問題，使得陣列中的元素不再共用一個快取行。
LongAccumulator
上面說到，LongAdder其實就是LongAccumulator的一個特例，相比LongAdder他的功能會更加強大，可以自定義累加的規則，在上面演示AtomicInteger功能的時候其實我們也使用過了。
*** ***，實際上就是實現了一個LongAdder的功能，初始值我們傳入0，而LongAdder的初始值就是0並且只能是0。
public class LongAdderTest {
public static void main(String[] args) {
LongAdder longAdder = new LongAdder();
LongAccumulator accumulator = new LongAccumulator((left, right) -> 0, 0);
}
}
工具類&容器類
這裡要說到一些我們在平時開發中經常使用到的一些類以及他們的實現原理。
CountDownLatch
CountDownLatch適用於在多執行緒的場景需要等待所有子執行緒全部執行完畢之後再做操作的場景。
假設現在我們有一個業務場景，我們需要呼叫多個RPC介面去查詢資料並且寫入excel，最後把所有excel打包壓縮傳送郵件出去。
public class CountDownLatchTest {
public static void main(String[] args) throws Exception{
ExecutorService executorService = Executors.newFixedThreadPool(10);
CountDownLatch countDownLatch = new CountDownLatch(2);
executorService.submit(()->{
try {
Thread.sleep(1000);
System.out.println("寫excelA完成");
countDownLatch.countDown();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
executorService.submit(()->{
try {
Thread.sleep(3000);
System.out.println("寫excelB完成");
countDownLatch.countDown();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
System.out.println("等待excel寫入完成");
countDownLatch.await();
System.out.println("開始打包傳送資料..");
executorService.shutdown();
}
}
//輸出
等待excel寫入完成
寫excelA完成
寫excelB完成
開始打包傳送資料..
整個過程如下：
初始化一個CountDownLatch範例傳參2，因為我們有2個子執行緒，每次子執行緒執行完畢之後呼叫countDown()方法給計數器-1，主執行緒呼叫await()方法後會被阻塞，直到最後計數器變為0，await()方法返回，執行完畢。
他和join有個區別，像我們這裡用的是ExecutorService建立執行緒池，是沒法使用join的，相比起來，CountDownLatch的使用會顯得更加靈活。
CountDownLatch基於AQS實現，用volatile修飾state變數維持倒數狀態，多執行緒共用變數可見。
- CountDownLatch通過建構函式初始化傳入引數實際為AQS的state變數賦值，維持計數器倒數狀態
- 當主執行緒呼叫await()方法時，當前執行緒會被阻塞，當state不為0時進入AQS阻塞佇列等待。
- 其他執行緒呼叫countDown()時，通過CAS修改state值-1，當state值為0的時候，喚醒所有呼叫await()方法阻塞的執行緒
CyclicBarrier
CyclicBarrier叫做迴環屏障，它的作用是讓一組執行緒全部達到一個狀態之後再全部同時執行，他和CountDownLatch主要區別在於，CountDownLatch的計數器只能用一次，而CyclicBarrier的計數器狀態則是可以一直重用的。
我們可以使用CyclicBarrier一樣實現上面的需求。
public class CyclicBarrierTest {
public static void main(String[] args) throws Exception{
ExecutorService executorService = Executors.newFixedThreadPool(10);
CyclicBarrier cyclicBarrier = new CyclicBarrier(2, () -> {
System.out.println("開始打包傳送資料..");
});
executorService.submit(()->{
try {
Thread.sleep(1000);
System.out.println("寫excelA完成");
cyclicBarrier.await();
} catch (Exception e) {
throw new RuntimeException(e);
}
});
executorService.submit(()->{
try {
Thread.sleep(3000);
System.out.println("寫excelB完成");
cyclicBarrier.await();
} catch (Exception e) {
throw new RuntimeException(e);
}
});
System.out.println("等待excel寫入完成");
executorService.shutdown();
}
}
//輸出
等待excel寫入完成
寫excelA完成
寫excelB完成
開始打包傳送資料..
初始化的時候我們傳入2個執行緒和一個回撥方法，執行緒呼叫await()之後進入阻塞狀態並且計數器-1，這個阻塞點被稱作為屏障點或者同步點，只有最後一個執行緒到達屏障點的時候，所有被屏障攔截的執行緒才能繼續執行，這也是叫做迴環屏障的名稱原因。
而當計數器為0時，就去執行CyclicBarrier建構函式中的回撥方法，回撥方法執行完成之後，就會退出屏障點，喚醒其他阻塞中的執行緒。
CyclicBarrier基於ReentrantLock實現，本質上還是基於AQS實現的，內部維護parties記錄匯流排程數，count用於計數，最開始count=parties，呼叫await()之後count原子遞減，當count為0之後，再次將parties賦值給count，這就是複用的原理。
- 當子執行緒呼叫await()方法時，獲取獨佔鎖ReentrantLock，同時對count遞減，進入阻塞佇列，然後釋放鎖
- 當第一個執行緒被阻塞同時釋放鎖之後，其他子執行緒競爭獲取鎖，操作同1
- 直到最後count為0，執行CyclicBarrier建構函式中的任務，執行完畢之後子執行緒繼續向下執行，計數重置，開始下一輪迴圈
Semaphore
Semaphore叫做號誌，和前面兩個不同的是，他的計數器是遞增的，號誌這玩意兒在限流中就經常使用到。
public class SemaphoreTest {
public static void main(String[] args) throws Exception {
ExecutorService executorService = Executors.newFixedThreadPool(10);
Semaphore semaphore = new Semaphore(0);
executorService.submit(() -> {
try {
Thread.sleep(1000);
System.out.println("寫excelA完成");
semaphore.release();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
executorService.submit(() -> {
try {
Thread.sleep(3000);
System.out.println("寫excelB完成");
semaphore.release();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
System.out.println("等待excel寫入完成");
semaphore.acquire(2);
System.out.println("開始打包傳送資料..");
executorService.shutdown();
}
}
//輸出
等待excel寫入完成
寫excelA完成
寫excelB完成
開始打包傳送資料..
稍微和前兩個有點區別，建構函式接受參數列示可用的許可證的數量，acquire方法表示獲取一個許可證，使用完之後release歸還許可證。
當子執行緒呼叫release()方法時，計數器遞增，主執行緒acquire()傳參為2則說明主執行緒一直阻塞，直到計數器為2才會返回。
Semaphore還還還是基於AQS實現的，同時獲取號誌有公平和非公平兩種策略，通過建構函式的傳參可以修改，預設則是非公平的策略。
- 先說非公平的策略，主執行緒呼叫acquire()方法時，用當前號誌值-需要獲取的值，如果小於0，說明還沒有達到號誌的要求值，則會進入AQS的阻塞佇列，大於0則通過CAS設定當前號誌為剩餘值，同時返回剩餘值。而對於公平策略來說，如果當前有其他執行緒在等待獲取資源，那麼自己就會進入AQS阻塞佇列排隊。
- 子執行緒呼叫release()給當前號誌值計數器+1(增加的值數量由傳參決定)，同時不停的嘗試喚醒因為呼叫acquire()進入阻塞的執行緒
Exchanger
Exchanger用於兩個執行緒之間交換資料，如果兩個執行緒都到達同步點，這兩個執行緒可以互相交換他們的資料。
舉個栗子，A和B兩個執行緒需要交換他們自己寫的資料以便核對資料是否一致。
public class ExchangerTest {
public static void main(String[] args) throws Exception {
ExecutorService executorService = Executors.newFixedThreadPool(10);
Exchanger<String> exchanger = new Exchanger<>();
executorService.submit(() -> {
try {
Thread.sleep(1000);
System.out.println("寫excelA完成");
System.out.println("A獲取到資料=" + exchanger.exchange("excelA"));
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
executorService.submit(() -> {
try {
Thread.sleep(3000);
System.out.println("寫excelB完成");
System.out.println("B獲取到資料=" + exchanger.exchange("excelB"));
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
executorService.shutdown();
}
}
//輸出
寫excelA完成
寫excelB完成
B獲取到資料=excelA
A獲取到資料=excelB
A寫完之後exchange會一直阻塞等待，直到另外一個執行緒也exchange之後，才會繼續執行。
ThreadLocalRandom
通常我們都會用 Random 去生成亂數，但是 Random 有點小問題，在多執行緒並行的情況下為了保證生成的隨機性，通過 CAS 的方式保證生成新種子的原子性，但是這樣帶來了效能的問題，多執行緒並行去生成亂數，但是隻有一個執行緒能成功，其他的執行緒會一直自旋，效能不高，所以 ThreadLocalRandom 就是為了解決這個問題而誕生。
//多執行緒下通過CAS保證新種子生成的原子性
protected int next(int bits) {
long oldseed, nextseed;
AtomicLong seed = this.seed;
do {
oldseed = seed.get();
nextseed = (oldseed * multiplier + addend) & mask;
} while (!seed.compareAndSet(oldseed, nextseed));
return (int)(nextseed >>> (48 - bits));
}
ThreadLocalRandom 我們從名字就能看出來，肯定使用了 ThreadLocal，作用就是用 ThreadLocal 儲存每個種子的變數，防止在高並行下對同一個種子的爭奪。
使用也非常簡單：
ThreadLocalRandom.current().nextInt(100);
看下原始碼實現，current 方法獲取當前的 ThreadLocalRandom 範例。
public static ThreadLocalRandom current() {
if (UNSAFE.getInt(Thread.currentThread(), PROBE) == 0)
localInit();
return instance;
}
nextInt 方法和 Random 看起來差不多，上面是生成新的種子，下面是固定的基於新種子計算亂數，主要看 nextSeed。
public int nextInt(int bound) {
if (bound <= 0)
throw new IllegalArgumentException(BadBound);
int r = mix32(nextSeed()); //生成新種子
int m = bound - 1;
if ((bound & m) == 0) // power of two
r &= m;
else { // reject over-represented candidates
for (int u = r >>> 1;
u + m - (r = u % bound) < 0;
u = mix32(nextSeed()) >>> 1)
;
}
return r;
}
r = UNSAFE.getLong(t, SEED) + GAMMA 計算出新的種子，然後使用 UNSAFE 的方法放入當前執行緒中。
final long nextSeed() {
Thread t; long r; // read and update per-thread seed
UNSAFE.putLong(t = Thread.currentThread(), SEED,
r = UNSAFE.getLong(t, SEED) + GAMMA);
return r;
}
ConcurrentHashMap
這個我們就不說了，說的太多了，之前的文章也寫過了，可以參考之前寫過的。
CopyOnWriteArrayList&CopyOnWriteArraySet
這是執行緒安全的 ArrayList ，從名字我們就能看出來，寫的時候複製，這叫做寫時複製，也就是寫的操作是對拷貝的陣列的操作。
先看建構函式，有3個，分別是無參，傳參為集合和傳引陣列，其實都差不多，無參建構函式建立一個新的陣列，集合則是把集合類的元素拷貝到新的陣列，陣列也是一樣。
public CopyOnWriteArrayList() {
setArray(new Object[0]);
}
public CopyOnWriteArrayList(Collection<? extends E> c) {
Object[] elements;
if (c.getClass() == CopyOnWriteArrayList.class)
elements = ((CopyOnWriteArrayList<?>)c).getArray();
else {
elements = c.toArray();
if (c.getClass() != ArrayList.class)
elements = Arrays.copyOf(elements, elements.length, Object[].class);
}
setArray(elements);
}
public CopyOnWriteArrayList(E[] toCopyIn) {
setArray(Arrays.copyOf(toCopyIn, toCopyIn.length, Object[].class));
}
我們看 add 方法，你一眼就能看出來非常簡單的實現，通過 ReentrantLock 加鎖，然後拷貝出一個新的陣列，陣列長度+1，再把新陣列賦值，所以這就是名字的由來，寫入的時候操作的是陣列的拷貝，其他的刪除修改就不看了，基本上是一樣的。
public boolean add(E e) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
Object[] elements = getArray();
int len = elements.length;
Object[] newElements = Arrays.copyOf(elements, len + 1);
newElements[len] = e;
setArray(newElements);
return true;
} finally {
lock.unlock();
}
}
再看看 get 方法，也非常簡單，直接獲取陣列當前索引的值，這裡需要注意的是，讀資料是沒有加鎖的，所以會有一致性的問題，它並不能保證讀到的一定是最新的資料。
public E get(int index) {
return get(getArray(), index);
}
private E get(Object[] a, int index) {
return (E) a[index];
}
final Object[] getArray() {
return array;
}
至於 CopyOnWriteArraySet ，他就是基於 CopyOnWriteArrayList 實現的，這裡我們不再贅述。
public CopyOnWriteArraySet() {
al = new CopyOnWriteArrayList<E>();
}
public boolean add(E e) {
return al.addIfAbsent(e);
}
public boolean addIfAbsent(E e) {
Object[] snapshot = getArray();
return indexOf(e, snapshot, 0, snapshot.length) >= 0 ? false :
addIfAbsent(e, snapshot);
}
Fork/Join
Fork/Join 是一個並行執行任務的框架，利用的分而治之的思想。
Fork 是把一個大的任務拆分成若干個小任務並行執行，Join 則是合併拆分的子任務的結果集，最終計算出大任務的結果。
所以整個 Fork/Join 的流程可以認為就是兩步：
- Fork 拆分任務，直到拆分到最小粒度不可拆分為止
- Join 計算結果，把每個子任務的結果進行合併
這裡我們需要介紹一下主要的幾個類：
ForkJoinTask：就是我們的分治任務的抽象類
RecursiveTask：繼承於 ForkJoinTask，用於計算有返回結果的任務
RecursiveAction： 繼承於 ForkJoinTask，用於計算沒有返回結果的任務
ForkJoinPool：用於執行 ForkJoinTask 任務的執行緒池，通常我們可以用 ForkJoinPool.commonPool() 去建立一個 Fork/Join 的執行緒池，然後用 submit 或者 invoke 去提交執行任務。
這裡我們寫一個測試程式，用於計算[0,999]的求和結果，所以我們寫一個類繼承 RecursiveTask ，並且實現他的 compute 方法。
invokeAll() 相當於每個任務都執行 fork，fork 之後會再次執行 compute 判斷是否要繼續拆分，如果無需拆分那麼則使用 join 方法計算彙總結果。
public class ForkJoinTest {
public static void main(String[] args) throws Exception {
List<Integer> list = new LinkedList<>();
Integer sum = 0;
for (int i = 0; i < 1000; i++) {
list.add(i);
sum += i;
}
CalculateTask task = new CalculateTask(0, list.size(), list);
Future<Integer> future = ForkJoinPool.commonPool().submit(task);
System.out.println("sum=" + sum + "，Fork/Join result=" + future.get());
}
@Data
static class CalculateTask extends RecursiveTask<Integer> {
private Integer start;
private Integer end;
private List<Integer> list;
public CalculateTask(Integer start, Integer end, List<Integer> list) {
this.start = start;
this.end = end;
this.list = list;
}
@Override
protected Integer compute() {
Integer sum = 0;
if (end - start < 200) {
for (int i = start; i < end; i++) {
sum += list.get(i);
}
} else {
int middle = (start + end) / 2;
System.out.println(String.format("從[%d,%d]拆分為：[%d,%d],[%d,%d]", start, end, start, middle, middle, end));
CalculateTask task1 = new CalculateTask(start, middle, list);
CalculateTask task2 = new CalculateTask(middle, end, list);
invokeAll(task1, task2);
sum = task1.join() + task2.join();
}
return sum;
}
}
}
//輸出
從[0,1000]拆分為：[0,500],[500,1000]
從[0,500]拆分為：[0,250],[250,500]
從[500,1000]拆分為：[500,750],[750,1000]
從[0,250]拆分為：[0,125],[125,250]
從[250,500]拆分為：[250,375],[375,500]
從[500,750]拆分為：[500,625],[625,750]
從[750,1000]拆分為：[750,875],[875,1000]
sum=499500，Fork/Join result=499500
使用完成之後，我們再來談一下 Fork/Join 的原理。
先看 fork 的程式碼，呼叫 fork 之後，使用workQueue.push() 把任務新增到佇列中，注意 push 之後呼叫 signalWork 喚醒一個執行緒去執行任務。
public final ForkJoinTask<V> fork() {
Thread t;
if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
((ForkJoinWorkerThread)t).workQueue.push(this);
else
ForkJoinPool.common.externalPush(this);
return this;
}
final ForkJoinPool.WorkQueue workQueue; // 工作竊取
final void push(ForkJoinTask<?> task) {
ForkJoinTask<?>[] a; ForkJoinPool p;
int b = base, s = top, n;
if ((a = array) != null) { // ignore if queue removed
int m = a.length - 1; // fenced write for task visibility
U.putOrderedObject(a, ((m & s) << ASHIFT) + ABASE, task);
U.putOrderedInt(this, QTOP, s + 1);
if ((n = s - b) <= 1) {
if ((p = pool) != null)
p.signalWork(p.workQueues, this);
}
else if (n >= m)
growArray();
}
}
上面我們看到了 workQueue，這個其實就是我們說的工作佇列，它是一個雙端佇列，並且有一個工作執行緒和他對應。
@sun.misc.Contended
static final class WorkQueue {
volatile int base; // 下一個出佇列索引
int top; // 下一個入佇列索引
ForkJoinTask<?>[] array; // 佇列中的 task
final ForkJoinPool pool;
final ForkJoinWorkerThread owner; // 工作佇列中的工作執行緒
volatile Thread parker; // == owner during call to park; else null
volatile ForkJoinTask<?> currentJoin; // 當前join的任務
volatile ForkJoinTask<?> currentSteal; // 當前偷到的任務
}
那如果工作執行緒自己佇列的做完了怎麼辦？只能傻傻地等待嗎？並不是，這時候有一個叫做工作竊取的機制，所以他就會去其他執行緒的佇列裡偷一個任務來執行。
為了避免偷任務執行緒和自己的執行緒產生競爭，所以自己的工作執行緒是從佇列頭部獲取任務執行，而偷任務執行緒則從佇列尾部偷任務。
Executor
Executor是並行程式設計中重要的一環，任務建立後提交到Executor執行並最終返回結果。
任務
執行緒兩種建立方式：Runnable和Callable。
Runnable是最初建立執行緒的方式，在JDK1.1的版本就已經存在，Callable則在JDK1.5版本之後加入，他們的主要區別在於Callable可以返回任務的執行結果。
任務執行
任務的執行主要靠Executor，ExecutorService繼承自Executor，ThreadPoolExecutor和ScheduledThreadPoolExecutor分別實現了ExecutorService。
那說到執行緒池之前，我們肯定要提及到執行緒池的幾個核心引數和原理，這個之前的文章也寫到過，屬於基礎中的基礎部分。
首先執行緒池有幾個核心的引數概念：
-
最大執行緒數maximumPoolSize
-
核心執行緒數corePoolSize
-
活躍時間keepAliveTime
-
阻塞佇列workQueue
-
拒絕策略RejectedExecutionHandler
當提交一個新任務到執行緒池時，具體的執行流程如下：
- 當我們提交任務，執行緒池會根據corePoolSize大小建立若干任務數量執行緒執行任務
- 當任務的數量超過corePoolSize數量，後續的任務將會進入阻塞佇列阻塞排隊
- 當阻塞佇列也滿了之後，那麼將會繼續建立(maximumPoolSize-corePoolSize)個數量的執行緒來執行任務，如果任務處理完成，maximumPoolSize-corePoolSize額外建立的執行緒等待keepAliveTime之後被自動銷燬
- 如果達到maximumPoolSize，阻塞佇列還是滿的狀態，那麼將根據不同的拒絕策略對應處理
拒絕策略主要有四種：
- AbortPolicy：直接丟棄任務，丟擲異常，這是預設策略
- CallerRunsPolicy：使用呼叫者所在的執行緒來處理任務
- DiscardOldestPolicy：丟棄等待佇列中最老的任務，並執行當前任務
- DiscardPolicy：直接丟棄任務，也不丟擲異常
ThreadPoolExecutor
通常為了快捷我們會用Executors工具類提供的建立執行緒池的方法快速地建立一個執行緒池出來，主要有幾個方法，但是一般我們不推薦這樣使用，非常容易導致出現問題，生產環境中我們一般推薦自己實現，引數自己定義，而不要使用這些方法。
建立
//建立固定執行緒數大小的執行緒池，核心執行緒數=最大執行緒數，阻塞佇列長度=Integer.MAX_VALUE
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
//建立只有一個執行緒的執行緒池，阻塞佇列長度=Integer.MAX_VALUE
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
//建立核心執行緒數為0，最大執行緒數=Integer.MAX_VALUE的執行緒池，阻塞佇列為同步佇列
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
最好的辦法就是自己建立，並且指定執行緒名稱：
new ThreadPoolExecutor(
Runtime.getRuntime().availableProcessors(),
Runtime.getRuntime().availableProcessors()*2,
1000L,
TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<>(100),
new ThreadFactoryBuilder().setNameFormat("thread-name").build());
提交任務
重點說一下幾個方法：
submit(Runnable task, T result)：可以用於主執行緒和子執行緒之間的通訊，資料共用。
submit(Runnable task)：返回null，相當於呼叫submit(Runnable task, null)。
invokeAll(Collection<? extends Callable
invokeAny(Collection<? extends Callable
public void execute(Runnable command); //提交runnable任務，無返回
public <T> Future<T> submit(Callable<T> task); //提交callable任務，有返回
public Future<?> submit(Runnable task); //提交runnable，有返回
public <T> Future<T> submit(Runnable task, T result); //提交runnable，有返回
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks); //批次提交任務
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit);
public <T> T invokeAny(Collection<? extends Callable<T>> tasks);
public <T> T invokeAny(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit);
關閉
shutdown：執行緒池狀態設定為SHUTDOWN，不再接受新任務，直接返回，執行緒池中任務會執行完成，遍歷執行緒池中的執行緒，逐個呼叫interrupt方法去中斷執行緒。
shutdownNow：執行緒池狀態設定為STOP，不再接受新任務，直接返回，執行緒池中任務會被中斷，返回值為被丟棄的任務列表。
isShutdown：只要呼叫了shutdown或者shutdownNow，都會返回true
isTerminating：所有任務都關閉後，才返回true
public void shutdown();
public List<Runnable> shutdownNow();
public boolean isShutdown();
public boolean isTerminating();
ScheduledThreadPoolExecutor
ScheduledThreadPoolExecutor 繼承於 ThreadPoolExecutor，從名字我們也知道，他是用於定時執行任務的執行緒池。
內部實現了一個DelayedWorkQueue作為任務的阻塞佇列，ScheduledFutureTask 作為排程的任務，儲存到佇列中。
我們先看下他的建構函式，4個建構函式都不支援傳佇列進來，所以預設的就是使用他的內部類 DelayedWorkQueue，由於 DelayedWorkQueue 是一個無界佇列，所以這裡最大執行緒數都是設定的為 Integer.MAX，因為沒有意義。
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue());
}
public ScheduledThreadPoolExecutor(int corePoolSize,
ThreadFactory threadFactory) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue(), threadFactory);
}
public ScheduledThreadPoolExecutor(int corePoolSize,
RejectedExecutionHandler handler) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue(), handler);
}
public ScheduledThreadPoolExecutor(int corePoolSize,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue(), threadFactory, handler);
}
執行定時任務的方法主要有4個，前面兩個 schedule 傳參區分 Runnable 和 Callable 其實並沒有區別，最終 Runnable 會通過 Executors.callable(runnable, result) 轉換為 Callable，本質上我們可以當做只有3個執行方法來看。
public <V> ScheduledFuture<V> schedule(Callable<V> callable,
long delay,
TimeUnit unit);
public ScheduledFuture<?> schedule(Runnable command,
long delay,
TimeUnit unit);
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
long initialDelay,
long period,
TimeUnit unit);
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
long initialDelay,
long delay,
TimeUnit unit);
schedule：提交一個延時任務，從時間單位為 unit 的 delay 時間開始執行，並且任務只會執行一次。
scheduleWithFixedDelay：以固定的延遲時間重複執行任務，initialDelay 表示提交任務後多長時間開始執行，delay 表示任務執行時間間隔。
scheduleAtFixedRate：以固定的時間頻率重複執行任務，指的是以起始時間開始，然後以固定的時間間隔重複執行任務，initialDelay 表示提交任務後多長時間開始執行，然後從 initialDelay + N * period執行。
這兩個特別容易搞混，很難理解到底是個啥意思，記住了。
scheduleAtFixedRate 是上次執行完成之後立刻執行，scheduleWithFixedDelay 則是上次執行完成+delay 後執行。
看個例子，兩個任務都會延遲1秒，然後以2秒的間隔開始重複執行，任務睡眠1秒的時間。
scheduleAtFixedRate 由於任務執行的耗時比時間間隔小，所以始終是以2秒的間隔在執行。
scheduleWithFixedDelay 因為任務耗時用了1秒，導致後面的時間間隔都成了3秒。
public class ScheduledThreadPoolTest {
public static void main(String[] args) throws Exception {
ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(10);
executorService.scheduleAtFixedRate(() -> {
try {
System.out.println("scheduleAtFixedRate=" + new SimpleDateFormat("HH:mm:ss").format(new Date()));
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}, 1000, 2000, TimeUnit.MILLISECONDS);
executorService.scheduleWithFixedDelay(() -> {
try {
System.err.println("scheduleWithFixedDelay=" + new SimpleDateFormat("HH:mm:ss").format(new Date()));
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}, 1000, 2000, TimeUnit.MILLISECONDS);
// executorService.shutdown();
}
}
//輸出
scheduleAtFixedRate=01:17:05
scheduleWithFixedDelay=01:17:05
scheduleAtFixedRate=01:17:07
scheduleWithFixedDelay=01:17:08
scheduleAtFixedRate=01:17:09
scheduleAtFixedRate=01:17:11
scheduleWithFixedDelay=01:17:11
scheduleAtFixedRate=01:17:13
scheduleWithFixedDelay=01:17:14
scheduleAtFixedRate=01:17:15
scheduleAtFixedRate=01:17:17
scheduleWithFixedDelay=01:17:17
scheduleAtFixedRate=01:17:19
scheduleWithFixedDelay=01:17:20
scheduleAtFixedRate=01:17:21
我們把任務耗時調整到超過時間間隔，比如改成睡眠3秒，觀察輸出結果。
scheduleAtFixedRate 由於任務執行的耗時比時間間隔長，按照規定上次任務執行結束之後立刻執行，所以變成以3秒的時間間隔執行。
scheduleWithFixedDelay 因為任務耗時用了3秒，導致後面的時間間隔都成了5秒。
scheduleWithFixedDelay=01:46:21
scheduleAtFixedRate=01:46:21
scheduleAtFixedRate=01:46:24
scheduleWithFixedDelay=01:46:26
scheduleAtFixedRate=01:46:27
scheduleAtFixedRate=01:46:30
scheduleWithFixedDelay=01:46:31
scheduleAtFixedRate=01:46:33
scheduleWithFixedDelay=01:46:36
scheduleAtFixedRate=01:46:36
OK，最後來說說實現原理：
- 首先我們通過呼叫 schedule 的幾個方法，把任務新增到 ScheduledThreadPoolExecutor 去執行
- 接收到任務之後，會通過請求引數的延遲時間計算出真正需要執行任務的時間，然後把任務封裝成 RunnableScheduledFuture
- 然後把封裝之後的任務新增到延遲佇列中，任務 ScheduledFutureTask 實現了 comparable 介面，把時間越小的任務放在佇列頭，如果時間一樣，則會通過 sequenceNumber 去比較，也就是執行時間相同，先提交的先執行
- 最後執行緒池會從延遲佇列中去獲取任務執行，如果是一次性的任務，執行之後刪除佇列中的任務，如果是重複執行的，則再次計算時間，然後把任務新增到延遲佇列中
CompletionService
記得上面我將 ThreadPoolExecutor 的方法嗎，其中有一個 invokeAny 的方法，批次提交任務，只要有一個完成了，就直接返回，而不用一直傻傻地等，他的實現就是使用了 CompletionService ，我給你看一段原始碼。
private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,
boolean timed, long nanos)
throws InterruptedException, ExecutionException, TimeoutException {
if (tasks == null)
throw new NullPointerException();
int ntasks = tasks.size();
if (ntasks == 0)
throw new IllegalArgumentException();
ArrayList<Future<T>> futures = new ArrayList<Future<T>>(ntasks);
ExecutorCompletionService<T> ecs = new ExecutorCompletionService<T>(this);
}
看到了吧，OK，在我們想試試使用這個類之前，我們先試試 invokeAny 好使不。
public class CompletionServiceTest {
private static final int TOTAL = 10;
private static ExecutorService executorService = Executors.newFixedThreadPool(TOTAL);
public static void main(String[] args) throws Exception {
testInvokeAny();
}
private static void testInvokeAny() throws Exception {
List<TestTask> taskList = new LinkedList<>();
for (int i = 0; i < TOTAL; i++) {
taskList.add(new TestTask(i));
}
String value = executorService.invokeAny(taskList, 60, TimeUnit.SECONDS);
System.out.println("get value = " + value);
executorService.shutdown();
}
static class TestTask implements Callable<String> {
private Integer index;
public TestTask(Integer index) {
this.index = index;
}
@Override
public String call() throws Exception {
long sleepTime = ThreadLocalRandom.current().nextInt(1000, 10000);
System.out.println("task-" + index + " sleep " + sleepTime + " Ms");
Thread.sleep(sleepTime);
return "task-" + index;
}
}
}
//輸出
task-7 sleep 3072 Ms
task-4 sleep 1186 Ms
task-3 sleep 6182 Ms
task-9 sleep 7411 Ms
task-0 sleep 1882 Ms
task-1 sleep 8274 Ms
task-2 sleep 4789 Ms
task-5 sleep 8894 Ms
task-8 sleep 7211 Ms
task-6 sleep 5959 Ms
get value = task-4
看到效果了吧，耗時最短的任務返回，整個流程就結束了，那我們試試自己用 CompletionService 來實現這個效果看看。
public static void main(String[] args) throws Exception {
// testInvokeAny();
testCompletionService();
}
private static void testCompletionService() {
CompletionService<String> completionService = new ExecutorCompletionService(executorService);
List<Future> taskList = new LinkedList<>();
for (int i = 0; i < TOTAL; i++) {
taskList.add(completionService.submit(new TestTask(i)));
}
String value = null;
try {
for (int i = 0; i < TOTAL; i++) {
value = completionService.take().get();
if (value != null) {
System.out.println("get value = " + value);
break;
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
taskList.forEach(task -> {
task.cancel(true);
});
}
executorService.shutdown();
}
//輸出
task-4 sleep 5006 Ms
task-1 sleep 4114 Ms
task-2 sleep 4865 Ms
task-5 sleep 1592 Ms
task-3 sleep 6190 Ms
task-7 sleep 2482 Ms
task-8 sleep 9405 Ms
task-9 sleep 8798 Ms
task-6 sleep 2040 Ms
task-0 sleep 2111 Ms
get value = task-5
效果是一樣的，我們只是實現了一個簡化版的 invokeAny 功能，使用起來也挺簡單的。
實現原理也挺簡單的，哪個任務先完成，就把他丟到阻塞佇列裡，這樣取任務結果的時候直接從佇列裡拿，肯定是拿到最新的那一個。
非同步結果
通常，我們都會用 FutureTask 來獲取執行緒非同步執行的結果，基於 AQS 實現。
這個沒有說太多的必要，看看幾個方法就行了。
public V get();
public V get(long timeout, TimeUnit unit);
public boolean cancel(boolean mayInterruptIfRunning);
get 會阻塞的獲取執行緒非同步執行的結果，一般不建議直接使用，最好是使用帶超時時間的 get 方法。
我們可以通過 cancel 方法去嘗試取消任務的執行，引數代表是否支援中斷，如果任務未執行，那麼可以直接取消，如果任務執行中，使用 cancel(true) 會嘗試中斷任務。
CompletableFuture
之前我們都在使用 Future，要麼只能用 get 方法阻塞，要麼就用 isDone 來判斷，JDK1.8 之後新增了 CompletableFuture 用於非同步程式設計，它針對 Future 的功能增加了回撥能力，可以幫助我們簡化非同步程式設計。
CompletableFuture 主要包含四個靜態方法去建立物件，主要區別在於 supplyAsync 返回計算結果，runAsync 不返回，另外兩個方法則是可以指定執行緒池，如果不指定執行緒池則預設使用 ForkJoinPool，預設執行緒數為CPU核數。
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier);
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor);
public static CompletableFuture<Void> runAsync(Runnable runnable);
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor);
下面看看他的那些噁心人的幾十個方法，我估計能瘋。
序列
序列就不用解釋了，A->B->C 按照順序執行，下一個任務必須等上一個任務執行完成才可以。
主要包含 thenApply、thenAccept、thenRun 和 thenCompose，以及他們對應的帶 async 的非同步方法。
為了方便記憶我們要記住，有 apply 的有傳參有返回值，帶 accept 的有傳參但是沒有返回值，帶 run 的啥也沒有，帶 compose 的會返回一個新的 CompletableFuture 範例。
public static void main(String[] args) throws Exception {
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(1000);
System.out.println(Thread.currentThread() + "工作完成");
return "supplyAsync";
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
CompletableFuture newFuture = future.thenApply((ret) -> {
System.out.println(Thread.currentThread() + "thenApply=>" + ret);
return "thenApply";
}).thenAccept((ret) -> {
System.out.println(Thread.currentThread() + "thenAccept=>" + ret);
}).thenRun(() -> {
System.out.println(Thread.currentThread() + "thenRun");
});
CompletableFuture<String> composeFuture = future.thenCompose((ret) -> {
System.out.println(Thread.currentThread() + "thenCompose=>" + ret);
return CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(1000);
System.out.println(Thread.currentThread() + "thenCompose工作完成");
return "thenCompose";
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
});
System.out.println(future.get());
System.out.println(newFuture.get());
System.out.println(composeFuture.get());
}
//輸出
Thread[ForkJoinPool.commonPool-worker-9,5,main]工作完成
Thread[ForkJoinPool.commonPool-worker-9,5,main]thenCompose=>supplyAsync
Thread[main,5,main]thenApply=>supplyAsync
Thread[main,5,main]thenAccept=>thenApply
Thread[main,5,main]thenRun
supplyAsync
null
Thread[ForkJoinPool.commonPool-worker-2,5,main]thenCompose工作完成
thenCompose
AND 聚合
這個意思是下一個任務執行必須等前兩個任務完成可以。
主要包含 thenCombine、thenAcceptBoth、runAfterBoth ，以及他們對應的帶 async 的非同步方法，區別和上面一樣。
public static void main(String[] args) throws Exception {
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(1000);
System.out.println(Thread.currentThread() + "A工作完成");
return "A";
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(2000);
System.out.println(Thread.currentThread() + "B工作完成");
return "B";
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
CompletableFuture newFuture = future.thenCombine(future2, (ret1, ret2) -> {
System.out.println(Thread.currentThread() + "thenCombine=>" + ret1 + "," + ret2);
return "thenCombine";
}).thenAcceptBoth(future2, (ret1, ret2) -> {
System.out.println(Thread.currentThread() + "thenAcceptBoth=>" + ret1 + "," + ret2);
}).runAfterBoth(future2, () -> {
System.out.println(Thread.currentThread() + "runAfterBoth");
});
System.out.println(future.get());
System.out.println(future2.get());
System.out.println(newFuture.get());
}
//輸出
Thread[ForkJoinPool.commonPool-worker-9,5,main]A工作完成
A
Thread[ForkJoinPool.commonPool-worker-2,5,main]B工作完成
B
Thread[ForkJoinPool.commonPool-worker-2,5,main]thenCombine=>A,B
Thread[ForkJoinPool.commonPool-worker-2,5,main]thenAcceptBoth=>thenCombine,B
Thread[ForkJoinPool.commonPool-worker-2,5,main]runAfterBoth
null
Or 聚合
Or 聚合代表只要多個任務中有一個完成了，就可以繼續下面的任務。
主要包含 applyToEither、acceptEither、runAfterEither ，以及他們對應的帶 async 的非同步方法，區別和上面一樣，不再舉例了。
回撥/例外處理
whenComplete、handle 代表執行完成的回撥，一定會執行，exceptionally 則是任務執行發生異常的回撥。
public static void main(String[] args) throws Exception {
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(1000);
int a = 1 / 0;
return "success";
} catch (Exception e) {
throw new RuntimeException(e);
}
});
CompletableFuture newFuture = future.handle((ret, exception) -> {
System.out.println(Thread.currentThread() + "handle exception=>" + exception.getMessage());
return "handle";
});
future.whenComplete((ret, exception) -> {
System.out.println(Thread.currentThread() + "whenComplete exception=>" + exception.getMessage());
});
CompletableFuture exceptionFuture = future.exceptionally((e) -> {
System.out.println(Thread.currentThread() + "exceptionally exception=>" + e.getMessage());
return "exception";
});
System.out.println("task future = " + future.get());
System.out.println("handle future = " + newFuture.get());
System.out.println("exception future = " + exceptionFuture.get());
}
//輸出
Thread[ForkJoinPool.commonPool-worker-9,5,main]exceptionally exception=>java.lang.RuntimeException: java.lang.ArithmeticException: / by zero
Thread[main,5,main]whenComplete exception=>java.lang.RuntimeException: java.lang.ArithmeticException: / by zero
Thread[ForkJoinPool.commonPool-worker-9,5,main]handle exception=>java.lang.RuntimeException: java.lang.ArithmeticException: / by zero
Exception in thread "main" java.util.concurrent.ExecutionException: java.lang.RuntimeException: java.lang.ArithmeticException: / by zero
at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1908)
at com.example.demo.CompletableFutureTest3.main(CompletableFutureTest3.java:31)
Caused by: java.lang.RuntimeException: java.lang.ArithmeticException: / by zero
at com.example.demo.CompletableFutureTest3.lambda$main$0(CompletableFutureTest3.java:13)
at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1604)
at java.util.concurrent.CompletableFuture$AsyncSupply.exec(CompletableFuture.java:1596)
at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1067)
at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1703)
at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:172)
Caused by: java.lang.ArithmeticException: / by zero
at com.example.demo.CompletableFutureTest3.lambda$main$0(CompletableFutureTest3.java:10)
... 6 more
阻塞佇列
並行程式設計中，佇列是其中不可缺少的一環，其實前面在說到執行緒池的時候，就已經提及到了阻塞佇列了，這裡我們要一起看看 JUC 包下提供的這些佇列。
阻塞佇列中的阻塞包含兩層意思：
- 插入的時候，如果阻塞佇列滿，插入元素阻塞
- 刪除/查詢的時候，如果阻塞佇列空，刪除/查詢元素阻塞
下面列出佇列的一些插入和刪除元素的方法，一個個來說：
add：向佇列尾部插入元素，插入成功返回 true，佇列滿則丟擲IllegalStateException("Queue full")異常
offer：向佇列尾部插入元素，佇列滿返回 false，否則返回 true，帶超時的則是會阻塞，達到超時時間後返回
put：向佇列尾部插入元素，佇列滿會一直阻塞
remove：刪除佇列頭部元素，刪除成功返回 true，佇列空則丟擲NoSuchElementException異常
poll：刪除佇列頭部元素，刪除成功返回佇列頭部元素，佇列空返回null，帶超時的則是會阻塞，達到超時時間後返回
take：刪除佇列頭部元素，佇列空會一直阻塞
element：查詢佇列頭部元素，並且返回，佇列空則丟擲NoSuchElementException異常
peek：查詢佇列頭部元素，並且返回
ArrayBlockingQueue
ArrayBlockingQueue 從名字就知道，基於陣列實現的有界阻塞佇列，基於AQS支援公平和非公平策略。
還是看建構函式吧，可以傳入初始陣列大小，一旦設定之後大小就不能改變了，傳參可以支援公平和非公平，最後一個建構函式可以支援傳入集合進行初始化，但是長度不能超過 capacity，否則丟擲ArrayIndexOutOfBoundsException異常。
public ArrayBlockingQueue(int capacity);
public ArrayBlockingQueue(int capacity, boolean fair);
public ArrayBlockingQueue(int capacity, boolean fair, Collection<? extends E> c);
這個其實在上面介紹 Condition 的時候我們就已經實現過他了，這裡就不再說了，可以參考上面 Condition 的部分。
LinkedBlockingQueue
LinkedBlockingQueue 基於連結串列實現的有界阻塞佇列。
使用無參建構函式則連結串列長度為 Integer.MAX_VALUE，另外兩個建構函式和 ArrayBlockingQueue 差不多。
public LinkedBlockingQueue();
public LinkedBlockingQueue(int capacity);
public LinkedBlockingQueue(Collection<? extends E> c);
我們可以看看 put 和 take 的原始碼。
- 首先加鎖中斷
- 然後判斷如果達到了佇列的最大長度，那麼就阻塞等待，否則就把元素插入到佇列的尾部
- 注意這裡和 ArrayBlockingQueue 有個區別，這裡再次做了一次判斷，如果佇列沒滿，喚醒因為 put 阻塞的執行緒，為什麼要做判斷，因為他們不是一把鎖
- 最後的邏輯是一樣的，notEmpty 喚醒
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
int c = -1;
Node<E> node = new Node<E>(e);
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
try {
while (count.get() == capacity) {
notFull.await();
}
enqueue(node);
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
}
private void enqueue(Node<E> node) {
// assert putLock.isHeldByCurrentThread();
// assert last.next == null;
last = last.next = node;
}
private void signalNotEmpty() {
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
notEmpty.signal();
} finally {
takeLock.unlock();
}
}
take的邏輯也是非常類似啊。
- 加鎖中斷
- 判斷佇列是不是空了，空了的話就阻塞等待，否則就從佇列移除一個元素
- 然後再次做一次判斷，佇列要是不空，就喚醒阻塞的執行緒
- 最後喚醒 notFull 的執行緒
public E take() throws InterruptedException {
E x;
int c = -1;
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();
try {
while (count.get() == 0) {
notEmpty.await();
}
x = dequeue();
c = count.getAndDecrement();
if (c > 1)
notEmpty.signal();
} finally {
takeLock.unlock();
}
if (c == capacity)
signalNotFull();
return x;
}
private E dequeue() {
// assert takeLock.isHeldByCurrentThread();
// assert head.item == null;
Node<E> h = head;
Node<E> first = h.next;
h.next = h; // help GC
head = first;
E x = first.item;
first.item = null;
return x;
}
private void signalNotFull() {
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {
notFull.signal();
} finally {
putLock.unlock();
}
}
PriorityBlockingQueue
PriorityBlockingQueue 是支援優先順序的無界阻塞佇列，預設排序按照自然排序升序排列。
幾個建構函式，無參建構函式初始容量為11，可以自定義，也可以在建立的時候傳入 comparator 自定義排序規則。
public PriorityBlockingQueue();
public PriorityBlockingQueue(int initialCapacity);
public PriorityBlockingQueue(int initialCapacity, Comparator<? super E> comparator);
public PriorityBlockingQueue(Collection<? extends E> c);
直接看 put 和 take 方法吧，後面都這樣，其他的就忽略好了，找到 put 之後，發現直接就是呼叫的 offer，那我們就直接看 offer 的實現。
- 首先還是加鎖，然後看當前元素個數是否達到了陣列的上限，到了就呼叫 tryGrow 去擴容。
- 看是否實現了 Comparator 介面，是的話就用 Comparator 去排序，否則就用 Comparable 去比較，如果兩個都沒有，會報錯
- notEmpty 喚醒，最後解鎖
public void put(E e) {
offer(e); // never need to block
}
public boolean offer(E e) {
if (e == null)
throw new NullPointerException();
final ReentrantLock lock = this.lock;
lock.lock();
int n, cap;
Object[] array;
while ((n = size) >= (cap = (array = queue).length))
tryGrow(array, cap);
try {
Comparator<? super E> cmp = comparator;
if (cmp == null)
siftUpComparable(n, e, array);
else
siftUpUsingComparator(n, e, array, cmp);
size = n + 1;
notEmpty.signal();
} finally {
lock.unlock();
}
return true;
}
這裡，我們要繼續關注一下這個擴容的邏輯，到底是怎麼處理的？程式碼不長，但是看著很方的樣子。
- 首先，先釋放鎖，因為下面用 CAS 處理，估計怕擴容時間太長阻塞的執行緒太多
- 然後 CAS 修改 allocationSpinLock 為1
- CAS 成功的話，進行擴容的邏輯，如果長度小於64就擴容一倍，否則擴容一半
- 之前我們說他無界，其實不太對，這裡就判斷是否超過了最大長度，MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8，判斷一下有可能會丟擲記憶體溢位異常
- 然後建立一個新的物件陣列，並且 allocationSpinLock 重新恢復為0
- 執行了一次 Thread.yield()，讓出 CPU，因為有可能其他執行緒正在擴容，讓大家爭搶一下
- 最後確保新的物件陣列建立成功了，也就是擴容是沒有問題的，再次加鎖，陣列拷貝，結束
private void tryGrow(Object[] array, int oldCap) {
lock.unlock(); // must release and then re-acquire main lock
Object[] newArray = null;
if (allocationSpinLock == 0 &&
UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset,
0, 1)) {
try {
int newCap = oldCap + ((oldCap < 64) ?
(oldCap + 2) : // grow faster if small
(oldCap >> 1));
if (newCap - MAX_ARRAY_SIZE > 0) { // possible overflow
int minCap = oldCap + 1;
if (minCap < 0 || minCap > MAX_ARRAY_SIZE)
throw new OutOfMemoryError();
newCap = MAX_ARRAY_SIZE;
}
if (newCap > oldCap && queue == array)
newArray = new Object[newCap];
} finally {
allocationSpinLock = 0;
}
}
if (newArray == null) // back off if another thread is allocating
Thread.yield();
lock.lock();
if (newArray != null && queue == array) {
queue = newArray;
System.arraycopy(array, 0, newArray, 0, oldCap);
}
}
take 的邏輯基本一樣，最多有個排序的邏輯在裡面，就不再多說了。
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
E result;
try {
while ( (result = dequeue()) == null)
notEmpty.await();
} finally {
lock.unlock();
}
return result;
}
DelayQueue
DelayQueue 是支援延時的無界阻塞佇列，這個在我們聊 ScheduledThreadPoolExecutor 也談到過，裡面也使用了延遲佇列，只不過是它自己的一個內部類，DelayQueue 內部其實使用 PriorityQueue 來實現。
DelayQueue 的用法是新增元素的時候可以設定一個延遲時間，當時間到了之後才能從佇列中取出來，使用 DelayQueue 中的物件必須實現 Delayed 介面，重寫 getDelay 和 compareTo 方法，就像這樣，那實現其實可以看 ScheduledThreadPoolExecutor 裡面是怎麼做的，這裡我就不管那麼多，示意一下就好了。
public class Test {
public static void main(String[] args) throws Exception {
DelayQueue<User> delayQueue = new DelayQueue<>();
delayQueue.put(new User(1, "a"));
}
@Data
@NoArgsConstructor
@AllArgsConstructor
static class User implements Delayed {
private Integer id;
private String username;
@Override
public long getDelay(TimeUnit unit) {
return 0;
}
@Override
public int compareTo(Delayed o) {
return 0;
}
}
}
我們可以看看他的屬性和建構函式，吶看到了吧，使用的 PriorityQueue，另外建構函式比較簡單了，不說了。
private final transient ReentrantLock lock = new ReentrantLock();
private final PriorityQueue<E> q = new PriorityQueue<E>();
private Thread leader = null;
private final Condition available = lock.newCondition();
public DelayQueue();
public DelayQueue(Collection<? extends E> c);
OK，沒啥毛病，這裡我們要先看 take 方法，不能先看 put，否則我覺得鬧不明白。
- 來第一步加鎖，如果頭結點是空的，也就是佇列是空的話，阻塞，沒啥好說的
- 反之佇列有東西，我們就要去取了嘛，但是這裡要看物件自己實現的 getDelay 方法獲得延遲的時間，如果延遲的時間小於0，那說明到時間了，可以執行了，poll 返回
- 第一次，leader 執行緒肯定是空的，執行緒阻塞 delay 的時間之後才開始執行，完全沒毛病，然後 leader 重新 置為 null
- 當 leader 不是 null 的時候，說明其他執行緒在操作了，所以阻塞等待喚醒
- 最後，leader 為 null，喚醒阻塞中的執行緒，解鎖
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
for (;;) {
E first = q.peek();
if (first == null)
available.await();
else {
long delay = first.getDelay(NANOSECONDS);
if (delay <= 0)
return q.poll();
first = null; // don't retain ref while waiting
if (leader != null)
available.await();
else {
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
available.awaitNanos(delay);
} finally {
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
if (leader == null && q.peek() != null)
available.signal();
lock.unlock();
}
}
然後再來看 put 就會簡單多了，put 還是直接呼叫的 offer，看 offer 方法。
這裡使用的是 PriorityQueue 的 offer 方法，其實和我們上面說到的 PriorityBlockingQueue 差不多，不再多說了，新增到佇列頭部之後，leader 置為 null，喚醒，結束了。
public boolean offer(E e) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
q.offer(e);
if (q.peek() == e) {
leader = null;
available.signal();
}
return true;
} finally {
lock.unlock();
}
}
SynchronousQueue&LinkedTransferQueue
為什麼這兩個放一起說呢。。。因為這原始碼真的不想在這裡說一遍，這倆原始碼可以單獨出一個專題來寫，長篇精悍文章不適合他他們，就簡單先了解下。
SynchronousQueue 是一個不儲存元素的阻塞佇列，每個 put 必須等待 take，否則不能繼續新增元素。
如果你還記得我們上面說到執行緒池的地方，newCachedThreadPool 預設就是使用的 SynchronousQueue。
他就兩個構造方法，你一看就知道，對吧，支援公平和非公平，當然你也別問預設是啥，問就是非公平。
public SynchronousQueue();
public SynchronousQueue(boolean fair);
主要靠內部抽象類 Transferer，他的實現主要有兩個，TransferQueue 和 TransferStack。
注意：如果是公平模式，使用的是 TransferQueue 佇列，非公平則使用 TransferStack 棧。
abstract static class Transferer<E> {
abstract E transfer(E e, boolean timed, long nanos);
}
LinkedTransferQueue 是連結串列組成的無界阻塞佇列，看他內部類就知道了，這是個連結串列實現。
static final class Node {
final boolean isData; // 標記生產者或者消費者
volatile Object item; // 值
volatile Node next; // 下一個節點
volatile Thread waiter;
}
LinkedBlockingDeque
LinkedBlockingDeque 是連結串列組成的雙向阻塞佇列，它支援從佇列的頭尾進行進行插入和刪除元素。
建構函式有3個，不傳初始容量就是 Integer 最大值。
public LinkedBlockingDeque() {
this(Integer.MAX_VALUE);
}
public LinkedBlockingDeque(int capacity);
public LinkedBlockingDeque(Collection<? extends E> c);
看下雙向連結串列的結構：
static final class Node<E> {
E item;
Node<E> prev;
Node<E> next;
Node(E x) {
item = x;
}
}
因為是雙向連結串列，所以比其他的佇列多了一些方法，比如 add、addFirst、addLast，add 其實就是 addLast，offer、put 也是類似。
我們可以區分看一下 putFirst 和 putLast ，主要區別就是 linkFirst 和 linkLast，分別去佇列頭部和尾部新增新節點，其他基本一致。
public void putFirst(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
Node<E> node = new Node<E>(e);
final ReentrantLock lock = this.lock;
lock.lock();
try {
while (!linkFirst(node))
notFull.await();
} finally {
lock.unlock();
}
}
public void putLast(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
Node<E> node = new Node<E>(e);
final ReentrantLock lock = this.lock;
lock.lock();
try {
while (!linkLast(node))
notFull.await();
} finally {
lock.unlock();
}
}
結尾
本次長篇內容參考書籍和檔案
- Java 並行程式設計的藝術
- Java 並行程式設計之美
- Java 並行程式設計實戰
- Java 8實戰
- 極客時間：Java 並行程式設計實戰
OK，本期內容到此結束，我是艾小仙，我們過兩個月再見。