concurrent包是基於AQS (AbstractQueuedSynchronizer)框架的,AQS框架借助於兩個類:
- Unsafe(提供CAS操作)
- LockSupport(提供park/unpark操作)
因此,LockSupport非常重要。
兩個重點
(1)操作對象
歸根結底,LockSupport.park()和LockSupport.unpark(Thread thread)調用的是Unsafe中的native代碼:
//LockSupport中 public static void park() { UNSAFE.park(false, 0L); }
//LockSupport中 public static void unpark(Thread thread) { if (thread != null) UNSAFE.unpark(thread); }
Unsafe類中的對應方法:
//park public native void park(boolean isAbsolute, long time); //unpack public native void unpark(Object var1);
park函數是將當前調用Thread阻塞,而unpark函數則是將指定線程Thread喚醒。
與Object類的wait/notify機制相比,park/unpark有兩個優點:
- 以thread為操作對象更符合阻塞線程的直觀定義
- 操作更精准,可以准確地喚醒某一個線程(notify隨機喚醒一個線程,notifyAll喚醒所有等待的線程),增加了靈活性。
(2)關於“許可”
在上面的文字中,我使用了阻塞和喚醒,是為了和wait/notify做對比。
-
其實park/unpark的設計原理核心是“許可”:park是等待一個許可,unpark是為某線程提供一個許可。
如果某線程A調用park,那么除非另外一個線程調用unpark(A)給A一個許可,否則線程A將阻塞在park操作上。 -
有一點比較難理解的,是unpark操作可以再park操作之前。
也就是說,先提供許可。當某線程調用park時,已經有許可了,它就消費這個許可,然后可以繼續運行。這其實是必須的。考慮最簡單的生產者(Producer)消費者(Consumer)模型:Consumer需要消費一個資源,於是調用park操作等待;Producer則生產資源,然后調用unpark給予Consumer使用的許可。非常有可能的一種情況是,Producer先生產,這時候Consumer可能還沒有構造好(比如線程還沒啟動,或者還沒切換到該線程)。那么等Consumer准備好要消費時,顯然這時候資源已經生產好了,可以直接用,那么park操作當然可以直接運行下去。如果沒有這個語義,那將非常難以操作。 -
但是這個“許可”是不能疊加的,“許可”是一次性的。
比如線程B連續調用了三次unpark函數,當線程A調用park函數就使用掉這個“許可”,如果線程A再次調用park,則進入等待狀態。
Unsafe.park和Unsafe.unpark的底層實現原理
在Linux系統下,是用的Posix線程庫pthread中的mutex(互斥量),condition(條件變量)來實現的。
mutex和condition保護了一個_counter的變量,當park時,這個變量被設置為0,當unpark時,這個變量被設置為1。
源碼:
每個Java線程都有一個Parker實例,Parker類是這樣定義的:
class Parker : public os::PlatformParker { private: volatile int _counter ; ... public: void park(bool isAbsolute, jlong time); void unpark(); ... } class PlatformParker : public CHeapObj<mtInternal> { protected: pthread_mutex_t _mutex [1] ; pthread_cond_t _cond [1] ; ... }
可以看到Parker類實際上用Posix的mutex,condition來實現的。
在Parker類里的_counter字段,就是用來記錄“許可”的。
- park 過程
當調用park時,先嘗試能否直接拿到“許可”,即_counter>0時,如果成功,則把_counter設置為0,並返回:
void Parker::park(bool isAbsolute, jlong time) { // Ideally we'd do something useful while spinning, such // as calling unpackTime(). // Optional fast-path check: // Return immediately if a permit is available. // We depend on Atomic::xchg() having full barrier semantics // since we are doing a lock-free update to _counter. if (Atomic::xchg(0, &_counter) > 0) return;
如果不成功,則構造一個ThreadBlockInVM,然后檢查_counter是不是>0,如果是,則把_counter設置為0,unlock mutex並返回:
ThreadBlockInVM tbivm(jt); if (_counter > 0) { // no wait needed _counter = 0; status = pthread_mutex_unlock(_mutex);
否則,再判斷等待的時間,然后再調用pthread_cond_wait函數等待,如果等待返回,則把_counter設置為0,unlock mutex並返回:
if (time == 0) { status = pthread_cond_wait (_cond, _mutex) ; } _counter = 0 ; status = pthread_mutex_unlock(_mutex) ; assert_status(status == 0, status, "invariant") ; OrderAccess::fence();
- unpark 過程
當unpark時,則簡單多了,直接設置_counter為1,再unlock mutex返回。如果_counter之前的值是0,則還要調用pthread_cond_signal喚醒在park中等待的線程:
void Parker::unpark() { int s, status ; status = pthread_mutex_lock(_mutex); assert (status == 0, "invariant") ; s = _counter; _counter = 1; if (s < 1) { if (WorkAroundNPTLTimedWaitHang) { status = pthread_cond_signal (_cond) ; assert (status == 0, "invariant") ; status = pthread_mutex_unlock(_mutex); assert (status == 0, "invariant") ; } else { status = pthread_mutex_unlock(_mutex); assert (status == 0, "invariant") ; status = pthread_cond_signal (_cond) ; assert (status == 0, "invariant") ; } } else { pthread_mutex_unlock(_mutex); assert (status == 0, "invariant") ; } }
