摘要:有了這個基礎,才能發揮作用,使得在節點取消和異常時能夠保證隊列在多線程下的完整性。
Doug Lea是JDK中concurrent工具包的作者,這位大神是誰可以自行google。
本文淺析ReentrantLock(可重入鎖)的原理
Lock接口Lock接口定義了這幾個方法:
lock()
用來獲取鎖,如果鎖已經被其他線程占有,則進行等待,直到搶占到鎖;該方法在發送異常時不會自動釋放鎖,所以在使用時需要在finall塊中釋放鎖;
tryLock()和tryLock(long time, TimeUnit unit)
嘗試獲得鎖,如果鎖已經被其他線程占有,返回false,成功獲取鎖返回true;該方法不會等待,立即返回;而帶有參數的tryLock在等待時長內拿到鎖返回true,超時或者沒拿到鎖返回false;帶參數的方法還支持響應中斷;
lockInterruptibly()
支持中斷的lock();
unlock()
釋放鎖;
newCondition()
新建Condition,Condition以后會分析;
ReentrantLock實現了Lock接口,ReentrantLock中有一個重要的成員變量,同步器sync繼承了AbstractQueuedSynchronizer簡稱AQS,我們先介紹AQS;
AQS用一個隊列(結構是一個FIFO隊列)來管理同步狀態,當線程獲取同步狀態失敗時,會將當前線程包裝成一個Node放入隊列,當前線程進入阻塞狀態;當同步狀態釋放時,會從隊列去出線程獲取同步狀態。
AQS里定義了head、tail、state,他們都是volatile修飾的,head指向隊列的第一個元素,tail指向隊列的最后一個元素,state表示了同步狀態,這個狀態非常重要,在ReentrantLock中,state為0的時候代表鎖被釋放,state為1時代表鎖已經被占用;
看下面代碼:
private static final Unsafe unsafe = Unsafe.getUnsafe(); private static final long stateOffset; private static final long headOffset; private static final long tailOffset; private static final long waitStatusOffset; private static final long nextOffset; static { try { stateOffset = unsafe.objectFieldOffset (AbstractQueuedSynchronizer.class.getDeclaredField("state")); headOffset = unsafe.objectFieldOffset (AbstractQueuedSynchronizer.class.getDeclaredField("head")); tailOffset = unsafe.objectFieldOffset (AbstractQueuedSynchronizer.class.getDeclaredField("tail")); waitStatusOffset = unsafe.objectFieldOffset (Node.class.getDeclaredField("waitStatus")); nextOffset = unsafe.objectFieldOffset (Node.class.getDeclaredField("next")); } catch (Exception ex) { throw new Error(ex); } }
這一段靜態初始化代碼初始了state、head、tail等變量的在內存中的偏移量;Unsafe類是sun.misc下的類,不屬于java標準。Unsafe讓java可以像C語言一樣操作內存指針,其中就提供了CAS的一些原子操作和park、unpark對線程掛起與恢復的操作;關于CAS是concurrent工具包的基礎,以后會多帶帶介紹,其主要作用就是在硬件級別提供了compareAndSwap的功能,從而實現了比較和交換的原子性操作。
AQS還有一個內部類叫Node,它將線程封裝,利用prev和next可以將Node串連成雙向鏈表,這就是一開始說的FIFO的結構;
ReentrantLock提供了公平鎖和非公平鎖,我們這里從非公平鎖分析AQS的應用;
Lock調用lock()方法時調用了AQS的lock()方法,我們來看這個非公平鎖NonfairSync的lock方法:
final void lock() { //首先調用CAS搶占同步狀態state,如果成功則將當前線程設置為同步器的獨占線程, //這也是非公平的體現,因為新來的線程沒有馬上加入隊列尾部,而是先嘗試搶占同步狀態。 if (compareAndSetState(0, 1)) setExclusiveOwnerThread(Thread.currentThread()); else //搶占同步狀態失敗,調用AQS的acquire acquire(1); }
瞄一眼acquire方法:
public final void acquire(int arg) { //在這里還是先試著搶占一下同步狀態 if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }
tryAcquire調用的是NonfairSync的實現,然后又調用了Sync的nonfairTryAcquire方法:
final boolean nonfairTryAcquire(int acquires) { final Thread current = Thread.currentThread(); int c = getState(); if (c == 0) { //和之前一樣,利用CAS搶占同步狀態,成功則設置當前線程為獨占線程并且返回true if (compareAndSetState(0, acquires)) { setExclusiveOwnerThread(current); return true; } } //如果當前線程已經是獨占線程,即當前線程已經獲得了同步狀態則將同步狀態state加1, //這里是可重入鎖的體現 else if (current == getExclusiveOwnerThread()) { int nextc = c + acquires; if (nextc < 0) // overflow throw new Error("Maximum lock count exceeded"); setState(nextc); return true; } //沒有搶占到同步狀態返回false return false; }
再看addWaiter方法:
private Node addWaiter(Node mode) { //新建一個Node,封裝了當前線程和模式,這里傳入的是獨占模式Node.EXCLUSIVE Node node = new Node(Thread.currentThread(), mode); // Try the fast path of enq; backup to full enq on failure Node pred = tail; //如果tail不為空就不需要初始化node隊列了 if (pred != null) { //將node作為隊列最后一個元素入列 node.prev = pred; if (compareAndSetTail(pred, node)) { pred.next = node; //返回新建的node return node; } } //如果tail為空則表示node隊列還沒有初始化,此時初始化隊列 enq(node); return node; }
瞄一眼enq方法:
private Node enq(final Node node) { //無限loop直到CAS成功,其他地方也大量使用了無限loop for (;;) { Node t = tail; if (t == null) { // Must initialize //隊列尾部為空,必須初始化,head初始化為一個空node,不包含線程,tail = head if (compareAndSetHead(new Node())) tail = head; } else { //隊列已經初始化,將當前node加在列尾 node.prev = t; //將當前node設置為tail,CAS操作,enqueue安全 if (compareAndSetTail(t, node)) { t.next = node; return t; } } } }
拿到新建的node后傳給acquireQueued方法:
final boolean acquireQueued(final Node node, int arg) { boolean failed = true; try { //標記是否中斷狀態 boolean interrupted = false; for (;;) { //拿到當前node的前驅 final Node p = node.predecessor(); //如果前驅正好為head,即當前線程在列首,馬上tryAcquire搶占同步狀態 if (p == head && tryAcquire(arg)) { //搶占成功后,將當前節點的thread、prev清空作為head setHead(node); p.next = null; // help GC 原來的head等待GC回收 failed = false; return interrupted; } //沒有搶占成功后,判斷是否要park if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } }
瞄一眼shouldParkAfterFailedAcquire方法:
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { int ws = pred.waitStatus; if (ws == Node.SIGNAL) //如果前驅node的狀態為SIGNAL,說明當前node可以park /* * This node has already set status asking a release * to signal it, so it can safely park. */ return true; if (ws > 0) { //如果前驅的狀態大于0說明前驅node的thread已經被取消 /* * Predecessor was cancelled. Skip over predecessors and * indicate retry. */ do { //從前驅node開始,將取消的node移出隊列 //當前節點之前的節點不會變化,所以這里可以更新prev,而且不必用CAS來更新。 node.prev = pred = pred.prev; } while (pred.waitStatus > 0); pred.next = node; } else { //前驅node狀態等于0或者為PROPAGATE(以后會介紹) //將前驅node狀態設置為SIGNAL,返回false,表示當前node暫不需要park, //可以再嘗試一下搶占同步狀態 /* * waitStatus must be 0 or PROPAGATE. Indicate that we * need a signal, but don"t park yet. Caller will need to * retry to make sure it cannot acquire before parking. */ compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } return false; }
看一下parkAndCheckInterrupt方法:
private final boolean parkAndCheckInterrupt() { //阻塞當前線程 LockSupport.park(this); //返回當前線程是否設置中斷標志,并清空中斷標志 return Thread.interrupted(); }
這里解釋一下為什么要保存一下中斷標志:中斷會喚醒被park的阻塞線程,但被park的阻塞線程不會響應中斷,所以這里保存一下中斷狀態并返回,如果狀態為true說明發生過中斷,會補發一次中斷,即調用interrupt()方法
在acquireQueued中發生異常時執行cancelAcquire:
private void cancelAcquire(Node node) { // Ignore if node doesn"t exist if (node == null) return; //清空node的線程 node.thread = null; // Skip cancelled predecessors //移除被取消的前繼node,這里只移動了node的prev,沒有改變next Node pred = node.prev; while (pred.waitStatus > 0) node.prev = pred = pred.prev; // predNext is the apparent node to unsplice. CASes below will // fail if not, in which case, we lost race vs another cancel // or signal, so no further action is necessary. //獲取前繼node的后繼node Node predNext = pred.next; // Can use unconditional write instead of CAS here. // After this atomic step, other Nodes can skip past us. // Before, we are free of interference from other threads. //設置當前node等待狀態為取消,其他線程檢測到取消狀態會移除它們 node.waitStatus = Node.CANCELLED; // If we are the tail, remove ourselves. if (node == tail && compareAndSetTail(node, pred)) { //如果當前node為tail,將前驅node設置為tail(CAS) //設置前驅node(即現在的tail)的后繼為null(CAS) //此時,如果中間有取消的node,將沒有引用指向它,將被GC回收 compareAndSetNext(pred, predNext, null); } else { // If successor needs signal, try to set pred"s next-link // so it will get one. Otherwise wake it up to propagate. int ws; if (pred != head && ((ws = pred.waitStatus) == Node.SIGNAL || (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) && pred.thread != null) { //如果當前node既不是head也不是tail,設置前繼node的后繼為當前node后繼 Node next = node.next; if (next != null && next.waitStatus <= 0) compareAndSetNext(pred, predNext, next); } else { //喚醒當前node后繼 unparkSuccessor(node); } //當前node的next設置為自己 //注意現在當前node的后繼的prev還指向當前node,所以當前node還未被刪除,prev是在移除取消節點時更新的 //這里就是為什么在前面要從后往前找可換新的node原因了,next會導致死循環 node.next = node; // help GC } }
畫圖描述解析一下cancelAcquire:
首先看如何跳過取消的前驅
這時,前驅被取消的node并沒有被移出隊列,前驅的前驅的next還指向前驅;
如果當前node是tail的情況:
這時,沒有任何引用指向當前node;
如果當前node既不是tail也不是head:
這時,當前node的前驅的next指向當前node的后繼,當前node的next指向自己,pre都沒有更新;
如果當前node是head的后繼:
這時,只是簡單的將當前node的next指向自己;
到這里,當線程搶占同步狀態的時候,會進入FIFO隊列等待同步狀態被釋放。在unlock()方法中調用了同步器的release方法;看一下release方法:
public final boolean release(int arg) { //判斷是否釋放同步狀態成功 if (tryRelease(arg)) { Node h = head; if (h != null && h.waitStatus != 0) //如果head不為null,且head的等待狀態不為0, //喚醒后繼node的線程 unparkSuccessor(h); return true; } return false; }
再來看一下tryRelease方法(在Sync類中實現):
protected final boolean tryRelease(int releases) { int c = getState() - releases; //當前thread不是獨占模式的那個線程,拋出異常 if (Thread.currentThread() != getExclusiveOwnerThread()) throw new IllegalMonitorStateException(); boolean free = false; if (c == 0) { //如果同步狀態state為0,釋放成功,將獨占線程設置為null free = true; setExclusiveOwnerThread(null); } //更新同步狀態state setState(c); return free; }
繼續看unparkSuccessor(喚醒后繼node的tread)方法:
private void unparkSuccessor(Node node) { /* * If status is negative (i.e., possibly needing signal) try * to clear in anticipation of signalling. It is OK if this * fails or if status is changed by waiting thread. */ int ws = node.waitStatus; if (ws < 0) //head的等待狀態為負數,設置head的等待狀態為0 compareAndSetWaitStatus(node, ws, 0); /* * Thread to unpark is held in successor, which is normally * just the next node. But if cancelled or apparently null, * traverse backwards from tail to find the actual * non-cancelled successor. */ Node s = node.next; if (s == null || s.waitStatus > 0) { //如果head的后繼node不存在或者后繼node等待狀態大于0(即取消) //從尾部往當前node迭代找到等待狀態為負數的node,unpark //因為會有取消的節點 s = null; for (Node t = tail; t != null && t != node; t = t.prev) if (t.waitStatus <= 0) s = t; } if (s != null) LockSupport.unpark(s.thread); }總結
介紹完ReentrantLock后,我們大體了解了AQS的工作原理。AQS主要就是使用了同步狀態和隊列實現了鎖的功能。有了CAS這個基礎,AQS才能發揮作用,使得在enqueue、dequeque、節點取消和異常時能夠保證隊列在多線程下的完整性。
文章版權歸作者所有,未經允許請勿轉載,若此文章存在違規行為,您可以聯系管理員刪除。
轉載請注明本文地址:http://specialneedsforspecialkids.com/yun/68433.html
摘要:有了這個基礎,才能發揮作用,使得在節點取消和異常時能夠保證隊列在多線程下的完整性。 Doug Lea是JDK中concurrent工具包的作者,這位大神是誰可以自行google。 本文淺析ReentrantLock(可重入鎖)的原理 Lock接口 showImg(https://segmentfault.com/img/bV2671?w=276&h=176); Lock接口定義了這幾個...
摘要:有了這個基礎,才能發揮作用,使得在節點取消和異常時能夠保證隊列在多線程下的完整性。 Doug Lea是JDK中concurrent工具包的作者,這位大神是誰可以自行google。 本文淺析ReentrantLock(可重入鎖)的原理 Lock接口 showImg(https://segmentfault.com/img/bV2671?w=276&h=176); Lock接口定義了這幾個...
摘要:當前線程已經獲取過這個鎖,則此時是重入,改變的計數即可,返回表示加鎖成功。的核心在于使用更新鎖的狀態,并利用一個同步隊列將獲取鎖失敗的線程進行排隊,當前驅節點解鎖后再喚醒后繼節點,是一個幾乎純實現的加鎖與解鎖。 簡介 Java 并發編程離不開鎖, Synchronized 是常用的一種實現加鎖的方式,使用比較簡單快捷。在 Java 中還有另一種鎖,即 Lock 鎖。 Lock 是一個接...
摘要:開始獲取鎖終于輪到出場了,的調用過程和完全一樣,同樣拿不到鎖,然后加入到等待隊列隊尾然后,在阻塞前需要把前驅結點的狀態置為,以確保將來可以被喚醒至此,的執行也暫告一段落了安心得在等待隊列中睡覺。 showImg(https://segmentfault.com/img/remote/1460000016012467); 本文首發于一世流云的專欄:https://segmentfault...
摘要:的主要功能和關鍵字一致,均是用于多線程的同步。而僅支持通過查詢當前線程是否持有鎖。由于和使用的是同一把可重入鎖,所以線程可以進入方法,并再次獲得鎖,而不會被阻塞住。公平與非公平公平與非公平指的是線程獲取鎖的方式。 1.簡介 可重入鎖ReentrantLock自 JDK 1.5 被引入,功能上與synchronized關鍵字類似。所謂的可重入是指,線程可對同一把鎖進行重復加鎖,而不會被阻...
閱讀 2815·2021-10-13 09:48
閱讀 3776·2021-10-13 09:39
閱讀 3586·2021-09-22 16:04
閱讀 1816·2021-09-03 10:48
閱讀 837·2021-08-03 14:04
閱讀 2358·2019-08-29 15:18
閱讀 3400·2019-08-26 12:19
閱讀 2869·2019-08-26 12:08