1. <ul id="0c1fb"></ul>

      <noscript id="0c1fb"><video id="0c1fb"></video></noscript>
      <noscript id="0c1fb"><listing id="0c1fb"><thead id="0c1fb"></thead></listing></noscript>

      99热在线精品一区二区三区_国产伦精品一区二区三区女破破_亚洲一区二区三区无码_精品国产欧美日韩另类一区

      RELATEED CONSULTING
      相關(guān)咨詢
      選擇下列產(chǎn)品馬上在線溝通
      服務(wù)時(shí)間:8:30-17:00
      你可能遇到了下面的問(wèn)題
      關(guān)閉右側(cè)工具欄

      新聞中心

      這里有您想知道的互聯(lián)網(wǎng)營(yíng)銷解決方案
      AQS源碼分析--jdk1.8

      JDK1.8

      ArrayList源碼分析--jdk1.8
      LinkedList源碼分析--jdk1.8
      HashMap源碼分析--jdk1.8
      AQS源碼分析--jdk1.8
      ReentrantLock源碼分析--jdk1.8

      成都創(chuàng)新互聯(lián)成都網(wǎng)站建設(shè)按需求定制設(shè)計(jì),是成都網(wǎng)站營(yíng)銷公司,為木托盤提供網(wǎng)站建設(shè)服務(wù),有成熟的網(wǎng)站定制合作流程,提供網(wǎng)站定制設(shè)計(jì)服務(wù):原型圖制作、網(wǎng)站創(chuàng)意設(shè)計(jì)、前端HTML5制作、后臺(tái)程序開(kāi)發(fā)等。成都網(wǎng)站制作熱線:13518219792

      AbstractQueuedSynchronizer概述

      ??1. AQS是一個(gè)基于FIFO隊(duì)列,可以用于構(gòu)建鎖或者其他相關(guān)同步裝置的基礎(chǔ)框架。
      ?2. AQS提供了雙向鏈表。
      ?3. AQS分為共享模式和獨(dú)占模式。
      ?4.AQS基于volatile內(nèi)存可見(jiàn)性和CAS原子性操作實(shí)現(xiàn)線程間通信操作。

      AbstractQueuedSynchronizer數(shù)據(jù)結(jié)構(gòu)

      ??數(shù)據(jù)結(jié)構(gòu)是集合的精華所在,數(shù)據(jù)結(jié)構(gòu)往往也限制了集合的作用和側(cè)重點(diǎn),了解各種數(shù)據(jù)結(jié)構(gòu)是我們分析源碼的必經(jīng)之路。
      ?AQS的數(shù)據(jù)結(jié)構(gòu)如下:雙向鏈表
      ?AQS源碼分析--jdk1.8
      ?AQS實(shí)現(xiàn)共享資源的訪問(wèn)控制基礎(chǔ):
      ? ??1.state字段,即同步器狀態(tài)字段。用于共享資源的訪問(wèn)控制
      ? ??2.CLH隊(duì)列,F(xiàn)IFO等待隊(duì)列,存放競(jìng)爭(zhēng)失敗的線程。通常CLH隊(duì)列是一個(gè)自旋隊(duì)列,AQS以阻塞的方式實(shí)現(xiàn)
      ? ??CLH隊(duì)列的使用:
      AQS源碼分析--jdk1.8

      CLH掃盲

      自旋鎖
      學(xué)習(xí)了解自旋鎖之前先回顧一下互斥鎖
      互斥鎖
      線程在獲取互斥鎖的時(shí)候,如果發(fā)現(xiàn)鎖已經(jīng)被其它線程占有,那么線程就會(huì)驚醒休眠,然后在適當(dāng)?shù)臅r(shí)機(jī)(比如喚醒)在獲取鎖。
      自旋鎖
      那么自旋鎖顧名思義就是“自旋”。就是當(dāng)一個(gè)線程在嘗試獲取鎖失敗之后,線程不會(huì)休眠或者掛起,而是一直在循環(huán)檢測(cè)鎖是否被其它線程釋放。
      區(qū)別
      互斥鎖就是開(kāi)始開(kāi)銷要大于自旋鎖。臨界區(qū)持鎖時(shí)間的大小并不會(huì)對(duì)互斥鎖的開(kāi)銷造成影響,而自旋鎖是死循環(huán)檢測(cè),加鎖全程消耗cpu,起始開(kāi)銷雖然低于互斥鎖,但是隨著持鎖時(shí)間,加鎖的開(kāi)銷是線性增長(zhǎng)。
      適用的情況
      互斥鎖用于臨界區(qū)持鎖時(shí)間比較長(zhǎng)的操作,比如下面這些情況都可以考慮

      臨界區(qū)有IO操作
      臨界區(qū)代碼復(fù)雜或者循環(huán)量大
      臨界區(qū)競(jìng)爭(zhēng)非常激烈
      單核處理器
      自旋鎖就主要用在臨界區(qū)持鎖時(shí)間非常短且CPU資源不緊張的情況下。當(dāng)遞歸調(diào)用時(shí)有可能造成死鎖。
      線程(節(jié)點(diǎn))隊(duì)列
      了解了自旋鎖之后,在學(xué)習(xí)ReentrantLock的時(shí)候,一個(gè)線程在等待鎖的時(shí)候會(huì)被封裝成一個(gè)Node節(jié)點(diǎn),然后加入一個(gè)隊(duì)列中并檢測(cè)前一個(gè)節(jié)點(diǎn)是否是頭節(jié)點(diǎn),并且嘗試獲取鎖,如果獲取鎖成功就返回,否則就阻塞。直到上一個(gè)節(jié)點(diǎn)釋放鎖并喚醒它。這樣看來(lái)似乎跟自旋沒(méi)什么掛鉤。這是因?yàn)锳QS里面的CLH隊(duì)列是CLH隊(duì)列鎖的一種變形。先來(lái)了解一下CLH隊(duì)列鎖
      CLH隊(duì)列鎖
      CLH(Craig, Landin, and Hagersten locks): 是一個(gè)自旋鎖,能確保無(wú)饑餓性,提供先來(lái)先服務(wù)的公平性。
      CLH鎖也是一種基于鏈表的可擴(kuò)展、高性能、公平的自旋鎖,申請(qǐng)線程只在本地變量上自旋,它不斷輪詢前驅(qū)的狀態(tài),如果發(fā)現(xiàn)前驅(qū)釋放了鎖就結(jié)束自旋。http://www.2cto.com/kf/201412/363574.html這篇文章中有比較詳細(xì)的圖解。
      AQS中的CLH隊(duì)列
      了解了自旋鎖與CLH隊(duì)列鎖之后,在學(xué)習(xí)AQS中的CLH隊(duì)列就比較簡(jiǎn)單了。AQS中的CLH隊(duì)列主要是對(duì)CLH隊(duì)列鎖改動(dòng)了兩個(gè)地方
      1.節(jié)點(diǎn)結(jié)構(gòu)上做出改變。CLH隊(duì)列鎖的節(jié)點(diǎn)包含一個(gè)布爾類型locked的字段。如果要獲取鎖,就將這個(gè)locked設(shè)置為true。然后就不停的輪訓(xùn)前驅(qū)節(jié)點(diǎn)的locked是否釋放了鎖(這個(gè)過(guò)程我們就叫做自旋)。AQS的CLH隊(duì)列在結(jié)構(gòu)上引入了頭節(jié)點(diǎn),尾節(jié)點(diǎn)。并且擁有一個(gè)前節(jié)點(diǎn)與下一個(gè)節(jié)點(diǎn)的引用。
      2.在等待獲取鎖的機(jī)制上由自旋改成了等待阻塞。
      MCS
      MSC與CLH最大的不同并不是鏈表是顯示還是隱式,而是線程自旋的規(guī)則不同:CLH是在前趨結(jié)點(diǎn)的locked域上自旋等待,而MSC是在自己的
      結(jié)點(diǎn)的locked域上自旋等待。正因?yàn)槿绱耍鉀Q了CLH在NUMA系統(tǒng)架構(gòu)中獲取locked域狀態(tài)內(nèi)存過(guò)遠(yuǎn)的問(wèn)題。

      AbstractQueuedSynchronizer源碼分析

      /*
        * 提供了一個(gè)基于FIFO隊(duì)列,可以用于構(gòu)建鎖或者其他相關(guān)同步裝置的基礎(chǔ)框架
       * 雙向鏈表
       */
      public abstract class AbstractQueuedSynchronizer
          extends AbstractOwnableSynchronizer
          implements java.io.Serializable {
      
         /**
           * 無(wú)參構(gòu)造方法
           */
          protected AbstractQueuedSynchronizer() { }
         /**
           * 
           *      +------+  prev +-----+       +-----+
           * head |      | <---- |     | <---- |     |  tail
           *      +------+       +-----+       +-----+
           * 
      */ static final class Node { /** Marker to indicate a node is waiting in shared mode 模式,分為共享與獨(dú)占 共享模式 */ static final Node SHARED = new Node(); /** Marker to indicate a node is waiting in exclusive mode 獨(dú)占模式 */ static final Node EXCLUSIVE = null; /** waitStatus value to indicate thread has cancelled * 結(jié)點(diǎn)狀態(tài) 節(jié)點(diǎn)watiStatus的值 * CANCELLED,值為1,終態(tài),該節(jié)點(diǎn)被取消由于超時(shí)或中斷 * SIGNAL,值為-1,表示當(dāng)前節(jié)點(diǎn)的后繼節(jié)點(diǎn)包含的線程需要運(yùn)行,也就是unpark,所以當(dāng)前節(jié)點(diǎn)release或cancels時(shí),必須unpark它的后繼節(jié)點(diǎn) * CONDITION,值為-2,表示當(dāng)前節(jié)點(diǎn)在等待condition,也就是在condition隊(duì)列中 該節(jié)點(diǎn)處于條件隊(duì)列中,將不會(huì)被用于sync queue,直到節(jié)點(diǎn)狀態(tài)被設(shè)置為0 * PROPAGATE,值為-3,表示當(dāng)前場(chǎng)景下后續(xù)的acquireShared能夠得以執(zhí)行releaseShared應(yīng)該被傳播到其他節(jié)點(diǎn) * 值為0,表示當(dāng)前節(jié)點(diǎn)在sync隊(duì)列中,等待著獲取鎖 * */ static final int CANCELLED = 1; /** waitStatus value to indicate successor's thread needs unparking */ static final int SIGNAL = -1; /** waitStatus value to indicate thread is waiting on condition */ static final int CONDITION = -2; /** * waitStatus value to indicate the next acquireShared should * unconditionally propagate */ static final int PROPAGATE = -3; /** * Status field, taking on only the values: * SIGNAL: The successor of this node is (or will soon be) * blocked (via park), so the current node must * unpark its successor when it releases or * cancels. To avoid races, acquire methods must * first indicate they need a signal, * then retry the atomic acquire, and then, * on failure, block. * CANCELLED: This node is cancelled due to timeout or interrupt. * Nodes never leave this state. In particular, * a thread with cancelled node never again blocks. * CONDITION: This node is currently on a condition queue. * It will not be used as a sync queue node * until transferred, at which time the status * will be set to 0. (Use of this value here has * nothing to do with the other uses of the * field, but simplifies mechanics.) * PROPAGATE: A releaseShared should be propagated to other * nodes. This is set (for head node only) in * doReleaseShared to ensure propagation * continues, even if other operations have * since intervened. * 0: None of the above * 結(jié)點(diǎn)狀態(tài) */ volatile int waitStatus; /** * 前驅(qū)結(jié)點(diǎn) */ volatile Node prev; /** * 后繼結(jié)點(diǎn) */ volatile Node next; /** * 結(jié)點(diǎn)所對(duì)應(yīng)的線程 */ volatile Thread thread; /** * 下一個(gè)等待者 */ Node nextWaiter; /** * 結(jié)點(diǎn)是否在共享模式下等待 */ final boolean isShared() { return nextWaiter == SHARED; } /** * 獲取前驅(qū)結(jié)點(diǎn),若前驅(qū)結(jié)點(diǎn)為空,拋出異常 */ final Node predecessor() throws NullPointerException { // 保存前驅(qū)結(jié)點(diǎn) Node p = prev; if (p == null) // 前驅(qū)結(jié)點(diǎn)為空,拋出異常 throw new NullPointerException(); else // 前驅(qū)結(jié)點(diǎn)不為空,返回 return p; } // 無(wú)參構(gòu)造函數(shù) Node() { // Used to establish initial head or SHARED marker } // 構(gòu)造函數(shù) Node(Thread thread, Node mode) { // Used by addWaiter this.nextWaiter = mode; this.thread = thread; } // 構(gòu)造函數(shù) Node(Thread thread, int waitStatus) { // Used by Condition this.waitStatus = waitStatus; this.thread = thread; } } /** * CLH隊(duì)列中頭結(jié)點(diǎn) */ private transient volatile Node head; /** * CLH隊(duì)列中尾結(jié)點(diǎn) */ private transient volatile Node tail; /** * 同步狀態(tài) * 多線程同步獲取資源成功,則state字段會(huì)自增;若有線程釋放資源,則state字段自減。 * 信號(hào)量 記錄該線程持有鎖的次數(shù)。 該線程每次釋放所 信號(hào)量 -1。 信號(hào)量為零 代表 鎖被真正釋放 */ private volatile int state; /** * @return current state value */ protected final int getState() { return state; } /** * @param newState the new state value */ protected final void setState(int newState) { state = newState; } /** * 使用unsafe的cas比較并且交換,保證原子性 */ protected final boolean compareAndSetState(int expect, int update) { // See below for intrinsics setup to support this return unsafe.compareAndSwapInt(this, stateOffset, expect, update); }

      AbstractQueuedSynchronizer繼承和實(shí)現(xiàn)分析

      AQS源碼分析--jdk1.8

      ?? AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer。
      ? 1. AbstractOwnableSynchronizer是一個(gè)抽象類一個(gè)同步器,它可能由線程獨(dú)占。該類為創(chuàng)建可能包含所有權(quán)概念的鎖和相關(guān)同步器提供了基礎(chǔ),但是,子類和工具可以使用適當(dāng)維護(hù)的值來(lái)幫助控制和監(jiān)視訪問(wèn)并提供診斷,實(shí)現(xiàn)了Serializable接口,定義了獨(dú)占模式,設(shè)置和獲取獨(dú)占模式下的線程Thread信息。
      ? 2.AbstractOwnableSynchronizer實(shí)現(xiàn)了Serializable接口。
      ? ??1)Serializable接口,序列化接口,表明該類可以被序列化,什么是序列化?簡(jiǎn)單的說(shuō),就是能夠從類變成字節(jié)流傳輸,反序列化,就是從字節(jié)流變成原來(lái)的類。
      ? 3. AbstractOwnableSynchronizer是一個(gè)抽象父類,子類有AbstractQueuedSynchronizer和AbstractQueuedLongSynchronizer,它們2個(gè)之間的區(qū)別就是異常將所有與狀態(tài)相關(guān)的參數(shù)和結(jié)果定義為long類型而不是int類型,在創(chuàng)建同步器(例如多級(jí)鎖和需要64位狀態(tài)的障礙)時(shí),此類可能很有用。??

      AbstractQueuedSynchronizer核心方法分析

      1. acquire方法--獨(dú)占模式

      ?? ??1)acquire(int arg);
      ? ??以獨(dú)占模式獲取資源,如果獲取成功,直接返回,否則進(jìn)去CLH等待隊(duì)列,通過(guò)自旋知道獲取到資源為止,過(guò)程中忽略線程中斷,獲取資源后才進(jìn)行自我中斷(補(bǔ)上),下面看源碼:

      /**
       * AQS的獨(dú)占模式--互斥
       * tryAcquire()嘗試直接去獲取資源,如果成功則直接返回;
       * addWaiter()將該線程加入等待隊(duì)列的尾部,并標(biāo)記為獨(dú)占模式;
       * acquireQueued()使線程在等待隊(duì)列中獲取資源,一直獲取到資源后才返回。如果在整個(gè)等待過(guò)程中被中斷過(guò),則返回true,否則返回false。
       * 如果線程在等待過(guò)程中被中斷過(guò),它是不響應(yīng)的。只是獲取資源后才再進(jìn)行自我中斷selfInterrupt(),將中斷補(bǔ)上。
       */
      public final void acquire(int arg) {
          if (!tryAcquire(arg) && // 再次嘗試上鎖 回到了  NonfairSync.tryAcquire 方法, tryAcquire 調(diào)用了 Sync.nonfairTryAcquire方法
              acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) // 鏈表尾部添加節(jié)點(diǎn) 為獨(dú)占模式
              selfInterrupt();
      }

      ?? ??2)boolean tryAcquire(int arg);
      ? ??嘗試以獨(dú)占的方式獲取資源,成功true,失敗false,該方法可以用于實(shí)現(xiàn)Lock中的tryLock()方法。

      /**
       * tryAcquire嘗試以獨(dú)占的方式獲取資源,如果獲取成功,則直接返回true,否則直接返回false。該方法可以用于實(shí)現(xiàn)Lock中的tryLock()方法。
       * 該方法的默認(rèn)實(shí)現(xiàn)是拋出UnsupportedOperationException,具體實(shí)現(xiàn)由自定義的擴(kuò)展了AQS的同步類來(lái)實(shí)現(xiàn)。AQS在這里只負(fù)責(zé)定義了一個(gè)公共的方法框架。
       * 這里之所以沒(méi)有定義成abstract,是因?yàn)楠?dú)占模式下只用實(shí)現(xiàn)tryAcquire-tryRelease,而共享模式下只用實(shí)現(xiàn)tryAcquireShared-tryReleaseShared。
       * 如果都定義成abstract,那么每個(gè)模式也要去實(shí)現(xiàn)另一模式下的接口
       * 由子類選擇性實(shí)現(xiàn)
       */
      protected boolean tryAcquire(int arg) {
          throw new UnsupportedOperationException();
      }

      ?? ??3)Node addWaiter(Node mode);
      ? ??將一個(gè)Node節(jié)點(diǎn)放入到CLH隊(duì)列的隊(duì)尾。

      /**
       * 將一個(gè)Node節(jié)點(diǎn)放入到CLH隊(duì)列的隊(duì)尾。
       * 第一步:首先將oldTail賦值給newNode.prev:node.prev = pred, 把當(dāng)前tail節(jié)點(diǎn)賦值到mode新節(jié)點(diǎn)的prev前一個(gè),
       * 第二步:將tail賦值給newNode:compareAndSetTail(pred, node) 把當(dāng)前tail節(jié)點(diǎn)的內(nèi)存地址修改為(指向)新的mode節(jié)點(diǎn),
       * 第三步:將oldTail的next指針指向newNode(即tail):pred.next = node 把當(dāng)前tail節(jié)點(diǎn)的next后一個(gè)賦值為新的mode節(jié)點(diǎn)(即tail)
       * 如果隊(duì)列為空,通過(guò)enq(node)方法初始化一個(gè)等待隊(duì)列,并返回當(dāng)前節(jié)點(diǎn)
       */
      private Node addWaiter(Node mode) {
          Node node = new Node(Thread.currentThread(), mode);
          // Try the fast path of enq; backup to full enq on failure
          //嘗試快速入隊(duì),失敗則使用enq()方式
          Node pred = tail;
          if (pred != null) { // 列隊(duì)尾部不為空
              node.prev = pred;
              if (compareAndSetTail(pred, node)) {
                  pred.next = node;
                  return node;
              }
          }
          // 列隊(duì)尾部為空 或者  CAS 操作失敗
          enq(node);
          return node;
      }

      ?? ??4)boolean acquireQueued(final Node node, int arg);
      ? ??使線程在等待隊(duì)列中獲取資源,一直獲取到資源后才返回。如果在整個(gè)等待過(guò)程中被中斷過(guò),則返回true,否則返回false。

      /**
       * 若node節(jié)點(diǎn)的前繼節(jié)點(diǎn)是head節(jié)點(diǎn),則會(huì)再次調(diào)用tryAcquire()獲取資源
       * 判斷當(dāng)前節(jié)點(diǎn)的前繼節(jié)點(diǎn)是否為head節(jié)點(diǎn)。若是,則表示該節(jié)點(diǎn)有資格嘗試獲取共享資源。此處的head節(jié)點(diǎn)的判斷在一定程度上保證資源競(jìng)爭(zhēng)的公平性
       * shouldParkAfterFailedAcquire():判斷當(dāng)前節(jié)點(diǎn)是否可以安全進(jìn)入park()
       * parkAndCheckInterrupt():讓線程進(jìn)入等待
       * 用于隊(duì)列中的線程自旋地以獨(dú)占且不可中斷的方式獲取同步狀態(tài)(acquire),直到拿到鎖之后再返回。該方法的實(shí)現(xiàn)分成兩部分:
       * 如果當(dāng)前節(jié)點(diǎn)已經(jīng)成為頭結(jié)點(diǎn),嘗試獲取鎖(tryAcquire)成功,然后返回;否則檢查當(dāng)前節(jié)點(diǎn)是否應(yīng)該被park,然后將該線程park并且檢查當(dāng)前線程是否被可以被中斷
       */
      final boolean acquireQueued(final Node node, int arg) {
          //標(biāo)記是否成功拿到資源,默認(rèn)false
          boolean failed = true;
          try {
              boolean interrupted = false;//標(biāo)記等待過(guò)程中是否被中斷過(guò)
              for (;;) {
                  final Node p = node.predecessor();
                  // 判斷當(dāng)前節(jié)點(diǎn)的 前驅(qū)節(jié)點(diǎn) 是否為隊(duì)列頭部  如果是 再次嘗試上鎖(如果頭部節(jié)點(diǎn) 已經(jīng)釋放鎖, 則使當(dāng)前線程成為持有者 并且設(shè)置自己為 頭部。 同時(shí)釋放前驅(qū)節(jié)點(diǎn))
                  if (p == head && tryAcquire(arg)) {
                      setHead(node);
                      p.next = null; // help GC
                      failed = false;
                      return interrupted;
                  }
                  //判斷當(dāng)前節(jié)點(diǎn)是否可以進(jìn)入park,若可以,讓線程進(jìn)入等待
                  if (shouldParkAfterFailedAcquire(p, node) &&
                      parkAndCheckInterrupt())
                      interrupted = true;
              }
          } finally {
              //如果獲取資源失敗,則取消
              if (failed)
                  cancelAcquire(node);
          }
      }

      ?? ??5)void selfInterrupt();
      ? ??中斷當(dāng)前線程

       /**
       * Convenience method to interrupt current thread.
       * 中斷當(dāng)前線程
       */
      static void selfInterrupt() {
          Thread.currentThread().interrupt();
      }

      ?? ??6)Node enq(final Node node);
      ? ??將當(dāng)前節(jié)點(diǎn)插入等待隊(duì)列

      /**
       * 進(jìn)行自旋入隊(duì)方式的enq()方法,基本和addWaiter()方法一致:
       * 用于將當(dāng)前節(jié)點(diǎn)插入等待隊(duì)列,如果隊(duì)列為空,則初始化當(dāng)前隊(duì)列。整個(gè)過(guò)程以CAS自旋的方式進(jìn)行,直到成功加入隊(duì)尾為止
       */
      private Node enq(final Node node) {
          for (;;) {
              Node t = tail;
              if (t == null) { // Must initialize 必須初始化 尾部為空 嘗試構(gòu)建表結(jié)構(gòu)
                  if (compareAndSetHead(new Node()))
                      tail = head;
              } else { //尾部不為空 不斷嘗試  CAS 操作
                  node.prev = t;
                  if (compareAndSetTail(t, node)) {
                      t.next = node;
                      return t;
                  }
              }
          }
      }

      ?? ??7)boolean compareAndSetHead(Node update);
      ? ??通過(guò)原子(CAS)操作 改變上鎖狀態(tài)

      /**
       * CAS head field. Used only by enq.
       * 通過(guò)原子操作 改變上鎖狀態(tài)
       * this == null
       * 第一個(gè)參數(shù)為需要改變的對(duì)象,第二個(gè)為偏移量(即之前求出來(lái)的valueOffset的值),第三個(gè)參數(shù)為期待的值,第四個(gè)為更新后的值
       */
      private final boolean compareAndSetHead(Node update) {
          //調(diào)用本地方法 實(shí)現(xiàn)硬件級(jí)別的原子操作 cas
          return unsafe.compareAndSwapObject(this, headOffset, null, update);
      }

      ?? ??8)boolean parkAndCheckInterrupt();
      ? ??讓線程去休息,真正進(jìn)入等待狀態(tài)
      ?

       /**
       * 該方法讓線程去休息,真正進(jìn)入等待狀態(tài)。park()會(huì)讓當(dāng)前線程進(jìn)入waiting狀態(tài)。在此狀態(tài)下,有兩種途徑可以喚醒該線程:
       * 1)被unpark();2)被interrupt()。需要注意的是,Thread.interrupted()會(huì)清除當(dāng)前線程的中斷標(biāo)記位
       */
      private final boolean parkAndCheckInterrupt() {
          LockSupport.park(this); // 又是一個(gè)底層類 實(shí)現(xiàn)線程等待
          return Thread.interrupted();
      }

      ?? ??9)boolean shouldParkAfterFailedAcquire(Node pred, Node node);
      ? ??判斷當(dāng)前節(jié)點(diǎn)中的線程,是否可以安全的進(jìn)入park()。返回true,表示進(jìn)程可以進(jìn)入park

      /**
       * 該方法的作用在于判斷當(dāng)前節(jié)點(diǎn)中的線程,是否可以安全的進(jìn)入park()。返回true,表示進(jìn)程可以進(jìn)入park。若前驅(qū)節(jié)點(diǎn)的waitStatus為SIGNAL,則表示當(dāng)前節(jié)點(diǎn)可以安全的park()。
       */
      private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
          int ws = pred.waitStatus;
          //如果前驅(qū)節(jié)點(diǎn)的waitStatus為SIGNAL -1,則表示當(dāng)前節(jié)點(diǎn)可以安全的park()
          if (ws == Node.SIGNAL)
              /*
               * This node has already set status asking a release
               * to signal it, so it can safely park.
               */
              return true;
          // waitStatus>0,即為CANCELLED狀態(tài),此時(shí)當(dāng)前節(jié)點(diǎn)需要找到狀態(tài)不為CANCELLED狀態(tài)的節(jié)點(diǎn),將其設(shè)置為自己的前驅(qū)節(jié)點(diǎn),并將新的前驅(qū)節(jié)點(diǎn)的next指向自己。
          // 注意,這樣做完之后,那些當(dāng)前節(jié)點(diǎn)的waitStatus狀態(tài)為CANCELLED的前驅(qū)節(jié)點(diǎn)鏈,將成為孤鏈。但這個(gè)孤鏈仍然有指向原等待隊(duì)列的prev和next指針。只是原等待隊(duì)列中已經(jīng)沒(méi)有指向孤鏈的節(jié)點(diǎn)指針
          // 將前驅(qū)節(jié)點(diǎn)移出列隊(duì)
          if (ws > 0) {
              /*
               * Predecessor was cancelled. Skip over predecessors and
               * indicate retry.
               */
              do {
                  node.prev = pred = pred.prev;
              } while (pred.waitStatus > 0);
              pred.next = node;
          } else {
              /*
               * 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.
               */
              // 走到此處,表明前驅(qū)節(jié)點(diǎn)的狀態(tài)為0或PROPAGATE。此時(shí)可以將前驅(qū)節(jié)點(diǎn)的waitStatus設(shè)置為SIGNAL狀態(tài)
              // 注意:這里仍然要返回false,表明當(dāng)前節(jié)點(diǎn)不能被park。我們需要在park之前,重試確認(rèn)該節(jié)點(diǎn)不能獲取到資源
              compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
          }
          return false;
      }       

      2. acquireShared方法--共享模式

      ?? ??1)acquireShared(int arg);
      ? ??以共享模式獲取資源,如果獲取成功,直接返回,否則進(jìn)去CLH等待隊(duì)列,通過(guò)自旋知道獲取到資源為止,過(guò)程中忽略線程中斷,獲取資源后才進(jìn)行自我中斷(補(bǔ)上),下面看源碼:

      /**
       *  aqs的共享模式
       *  獲取指定量的資源,獲取成功則直接返回,獲取失敗則進(jìn)入等待隊(duì)列,直到獲取到資源為止,整個(gè)過(guò)程忽略中斷
       */
      public final void acquireShared(int arg) {
          if (tryAcquireShared(arg) < 0)
              doAcquireShared(arg);
      }

      ?? ??2)int tryAcquireShared(int arg);
      ? ??嘗試以共享的方式獲取資源,成功true,失敗false,該方法可以用于實(shí)現(xiàn)Lock中的tryLock()方法。

      /**
       * tryAcquireShared嘗試以共享的方式獲取資源,如果獲取成功,則直接返回true,否則直接返回false。該方法可以用于實(shí)現(xiàn)Lock中的tryLock()方法。
       * 該方法的默認(rèn)實(shí)現(xiàn)是拋出UnsupportedOperationException,具體實(shí)現(xiàn)由自定義的擴(kuò)展了AQS的同步類來(lái)實(shí)現(xiàn)。AQS在這里只負(fù)責(zé)定義了一個(gè)公共的方法框架。
       * 這里之所以沒(méi)有定義成abstract,是因?yàn)楠?dú)占模式下只用實(shí)現(xiàn)tryAcquire-tryRelease,而共享模式下只用實(shí)現(xiàn)tryAcquireShared-tryReleaseShared。
       * 如果都定義成abstract,那么每個(gè)模式也要去實(shí)現(xiàn)另一模式下的接口
       */
      protected int tryAcquireShared(int arg) {
          throw new UnsupportedOperationException();
      }

      ?? ??3)doAcquireShared(int arg);
      ? ??將當(dāng)前線程加入等待隊(duì)列尾部休息,直到其他線程釋放資源喚醒自己,自己成功拿到相應(yīng)量的資源后才返回

      /**
       * Acquires in shared uninterruptible mode.
       * @param arg the acquire argument
       * 將當(dāng)前線程加入等待隊(duì)列尾部休息,直到其他線程釋放資源喚醒自己,自己成功拿到相應(yīng)量的資源后才返回
       */
      private void doAcquireShared(int arg) {
          final Node node = addWaiter(Node.SHARED);
          boolean failed = true;
          try {
              boolean interrupted = false;
              for (;;) {
                  final Node p = node.predecessor();
                  if (p == head) {
                      int r = tryAcquireShared(arg);
                      if (r >= 0) {
                          setHeadAndPropagate(node, r);
                          p.next = null; // help GC
                          if (interrupted)
                              selfInterrupt();
                          failed = false;
                          return;
                      }
                  }
                  if (shouldParkAfterFailedAcquire(p, node) &&
                      parkAndCheckInterrupt())
                      interrupted = true;
              }
          } finally {
              if (failed)
                  cancelAcquire(node);
          }
      }

      ?? ??4)boolean parkAndCheckInterrupt();
      ? ??讓線程去休息,真正進(jìn)入等待狀態(tài)

      /**
       * 該方法讓線程去休息,真正進(jìn)入等待狀態(tài)。park()會(huì)讓當(dāng)前線程進(jìn)入waiting狀態(tài)。在此狀態(tài)下,有兩種途徑可以喚醒該線程:
       * 1)被unpark();2)被interrupt()。需要注意的是,Thread.interrupted()會(huì)清除當(dāng)前線程的中斷標(biāo)記位
       */
      private final boolean parkAndCheckInterrupt() {
          LockSupport.park(this); // 又是一個(gè)底層類 實(shí)現(xiàn)線程等待
          return Thread.interrupted();
      }

      ?? ??5)cancelAcquire(Node node);
      ? ??取消節(jié)點(diǎn)

      /**
       * 取消節(jié)點(diǎn)
       * 列隊(duì)等待中 拋出異常會(huì)調(diào)用此方法
       */
      private void cancelAcquire(Node node) {
          // Ignore if node doesn't exist
          if (node == null)
              return;
      
          //找到適合的前繼節(jié)點(diǎn),當(dāng)前節(jié)點(diǎn)的waitStatus賦值為CANCELLED
          node.thread = null; // 釋放線程
      
          // Skip cancelled predecessors 前驅(qū)節(jié)點(diǎn)已被取消  重新定義前驅(qū)節(jié)點(diǎn)
          Node pred = node.prev;
          //若前繼節(jié)點(diǎn)是CANCELLED,則繼續(xù)找前繼節(jié)點(diǎn),直至找到一個(gè)正常的前繼節(jié)點(diǎn)賦值給node,作為node的新前繼節(jié)點(diǎn)
          while (pred.waitStatus > 0)
              node.prev = pred = pred.prev;
      
          Node predNext = pred.next;
      
          node.waitStatus = Node.CANCELLED; // 取消當(dāng)前線程 所屬的節(jié)點(diǎn)(標(biāo)記為取消),  沒(méi)有使用 cas  因?yàn)?其他線程 不會(huì)干擾這里
      
          // If we are the tail, remove ourselves.
          //特殊情況:node==tail節(jié)點(diǎn),將pred作為tail節(jié)點(diǎn),然后將cancelledNodes節(jié)點(diǎn)鏈從CLH隊(duì)列剔除
          if (node == tail && compareAndSetTail(node, pred)) {
              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;
              //正常情況:則將cancelledNodes節(jié)點(diǎn)鏈從CLH隊(duì)列剔除
              if (pred != head &&
                  ((ws = pred.waitStatus) == Node.SIGNAL ||
                   (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
                  pred.thread != null) {
                  Node next = node.next;
                  if (next != null && next.waitStatus <= 0)
                      compareAndSetNext(pred, predNext, next);
              } else {
                  //特殊情況:如果node是head的后繼節(jié)點(diǎn),則直接喚醒node的后繼節(jié)點(diǎn) pred==head節(jié)點(diǎn):嘗試調(diào)用unparkSuccessor(node),嘗試喚醒當(dāng)前節(jié)點(diǎn)的后繼節(jié)點(diǎn)
                  unparkSuccessor(node);
              }
      
              node.next = node; // help GC
          }
      }

      ?? ??6)unparkSuccessor(Node node);
      ? ??喚醒后繼節(jié)點(diǎn)

          /**
       * 喚醒后繼節(jié)點(diǎn)
       * 注意:如果當(dāng)前節(jié)點(diǎn)的后繼節(jié)點(diǎn)為空,或者是被取消的節(jié)點(diǎn)。那就從tail節(jié)點(diǎn)逆向遍歷CLH隊(duì)列,直至找到一個(gè)距離當(dāng)前節(jié)點(diǎn)node最近,且waitStatus<=0的節(jié)點(diǎn),然后喚醒該節(jié)點(diǎn)
       */
      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; //獲取頭結(jié)點(diǎn)的等待狀態(tài)
          if (ws < 0) //把該狀態(tài)設(shè)置成0
              compareAndSetWaitStatus(node, ws, 0);
      
          /*
           * 若后繼節(jié)點(diǎn)不符合喚醒標(biāo)準(zhǔn),則逆向遍歷CLH,直至找到一個(gè)距離當(dāng)前節(jié)點(diǎn)node最近,且waitStatus<=0的節(jié)點(diǎn)
           */
          Node s = node.next; //找到后繼節(jié)點(diǎn),喚醒后繼節(jié)點(diǎn)
          if (s == null || s.waitStatus > 0) { //很不巧,后繼節(jié)點(diǎn),節(jié)點(diǎn)為null,或者被取消
              s = null;
              for (Node t = tail; t != null && t != node; t = t.prev) //這里采用反向遍歷因?yàn)槭请p向鏈表
                  if (t.waitStatus <= 0) //找到實(shí)際未被取消的節(jié)點(diǎn)
                      s = t;
          }
          if (s != null)
              LockSupport.unpark(s.thread); //喚醒節(jié)點(diǎn)
      }

      3. release方法--釋放資源

      ?? ??1)boolean release(int arg);
      ? ??獨(dú)占模式釋放資源

      /**
       * 資源的釋放
       * 調(diào)用tryRelease方法進(jìn)行釋放鎖
       * 釋放鎖成功后,獲取頭節(jié)點(diǎn),接著喚醒后繼節(jié)點(diǎn),調(diào)用unparkSuccessor方法
       */
      public final boolean release(int arg) {
          if (tryRelease(arg)) {
              Node h = head;
              if (h != null && h.waitStatus != 0)
                  unparkSuccessor(h);
              return true;
          }
          return false;
      }

      ?? ??2)boolean tryRelease(int arg);
      ? ??獨(dú)占模式下嘗試釋放鎖,由子類選擇性實(shí)現(xiàn)

              protected boolean tryRelease(int arg) {
                      throw new UnsupportedOperationException();
              }

      ?? ??3)boolean releaseShared(int arg);
      ? ??共享模式釋放資源

      /**
       * Releases in shared mode.  Implemented by unblocking one or more
       * threads if {@link #tryReleaseShared} returns true.
       *
       * @param arg the release argument.  This value is conveyed to
       *        {@link #tryReleaseShared} but is otherwise uninterpreted
       *        and can represent anything you like.
       * @return the value returned from {@link #tryReleaseShared}
       */
      public final boolean releaseShared(int arg) {
          if (tryReleaseShared(arg)) {
              doReleaseShared();
              return true;
          }
          return false;
      }

      ?? ??4)boolean tryReleaseShared(int arg);
      ? ??共享模式下嘗試釋放鎖,由子類選擇性實(shí)現(xiàn)

          protected boolean tryReleaseShared(int arg) {
                      throw new UnsupportedOperationException();
              }

      ?? ??5)int fullyRelease(Node node);
      ? ??使用當(dāng)前節(jié)點(diǎn)狀態(tài)調(diào)用release,成功返回狀態(tài),失敗跑出異常

      /**
       * 使用當(dāng)前節(jié)點(diǎn)狀態(tài)調(diào)用release,成功返回狀態(tài),失敗跑出異常
       */
      final int fullyRelease(Node node) {
          boolean failed = true;
          try {
              int savedState = getState();
              if (release(savedState)) {
                  failed = false;
                  return savedState;
              } else {
                  throw new IllegalMonitorStateException();
              }
          } finally {
              if (failed)
                  node.waitStatus = Node.CANCELLED;
          }
      }

      4. CAS操作

       /**
       * Unsafe類實(shí)例
       */
      private static final Unsafe unsafe = Unsafe.getUnsafe();
      /**  state內(nèi)存偏移地址 */
      private static final long stateOffset;
      /**  head內(nèi)存偏移地址 */
      private static final long headOffset;
      /** tail內(nèi)存偏移地址 */
      private static final long tailOffset;
      private static final long waitStatusOffset;
      /** next內(nèi)存偏移地址 */
      private static final long nextOffset;
      
      //靜態(tài)初始化塊
      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); }
      }
      
      /**
       * CAS head field. Used only by enq.
       * 通過(guò)原子操作 改變上鎖狀態(tài)
       * this == null
       * 第一個(gè)參數(shù)為需要改變的對(duì)象,第二個(gè)為偏移量(即之前求出來(lái)的valueOffset的值),第三個(gè)參數(shù)為期待的值,第四個(gè)為更新后的值
       */
      private final boolean compareAndSetHead(Node update) {
          //調(diào)用本地方法 實(shí)現(xiàn)硬件級(jí)別的原子操作 cas
          return unsafe.compareAndSwapObject(this, headOffset, null, update);
      }
      
      /**
       * CAS tail field. Used only by enq.
       */
      private final boolean compareAndSetTail(Node expect, Node update) {
          return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
      }
      
      /**
       * CAS waitStatus field of a node.
       */
      private static final boolean compareAndSetWaitStatus(Node node,
                                                           int expect,
                                                           int update) {
          return unsafe.compareAndSwapInt(node, waitStatusOffset,
                                          expect, update);
      }
      
      /**
       * CAS next field of a node.
       */
      private static final boolean compareAndSetNext(Node node,
                                                     Node expect,
                                                     Node update) {
          return unsafe.compareAndSwapObject(node, nextOffset, expect, update);
      }

      5. Condition條件隊(duì)列

      內(nèi)部類ConditionObject,它實(shí)現(xiàn)了Condition接口,主要用于實(shí)現(xiàn)條件鎖。
      ConditionObject中也維護(hù)了一個(gè)隊(duì)列,這個(gè)隊(duì)列主要用于等待條件的成立,當(dāng)條件成立時(shí),其它線程將signal這個(gè)隊(duì)列中的元素,將其移動(dòng)到CLH的隊(duì)列中,等待占有鎖的線程釋放鎖后被喚醒。
      Condition典型的運(yùn)用場(chǎng)景是在BlockingQueue中的實(shí)現(xiàn),當(dāng)隊(duì)列為空時(shí),獲取元素的線程阻塞在notEmpty條件上,一旦隊(duì)列中添加了一個(gè)元素,將通知notEmpty條件,將其隊(duì)列中的元素移動(dòng)到AQS隊(duì)列中等待被喚醒。

      /**
       * 構(gòu)造一個(gè)條件隊(duì)列,來(lái)等待條件是否為真
       */
      public class ConditionObject implements Condition, java.io.Serializable {
          /** 版本號(hào) */
          private static final long serialVersionUID = 1173984872572414699L;
          /** First node of condition queue. condition隊(duì)列的頭結(jié)點(diǎn) */
          private transient Node firstWaiter;
          /** Last node of condition queue. condition隊(duì)列的尾結(jié)點(diǎn) */
          private transient Node lastWaiter;
      
          /**
           * Creates a new {@code ConditionObject} instance.
           * 構(gòu)造函數(shù)
           */
          public ConditionObject() { }
      
          /**
           * Adds a new waiter to wait queue.
           * @return its new wait node
           * 添加新的waiter到wait隊(duì)列
           */
          private Node addConditionWaiter() {
              //保存尾結(jié)點(diǎn)
              Node t = lastWaiter;
              // If lastWaiter is cancelled, clean out. 尾結(jié)點(diǎn)不為空,并且尾結(jié)點(diǎn)的狀態(tài)不為CONDITION
              if (t != null && t.waitStatus != Node.CONDITION) {
                  //清除狀態(tài)為CONDITION的結(jié)點(diǎn)
                  unlinkCancelledWaiters();
                  //將最后一個(gè)結(jié)點(diǎn)重新賦值給t
                  t = lastWaiter;
              }
              Node node = new Node(Thread.currentThread(), Node.CONDITION);
              if (t == null)
                  firstWaiter = node;
              else
                  t.nextWaiter = node;
              lastWaiter = node;
              return node;
          }

      AQS總結(jié)

      1)AQS分為獨(dú)占鎖和共享鎖。
      2)AQS分為CLH自旋隊(duì)列和Condition條件隊(duì)列。
      3)AQS是一個(gè)雙向鏈表,由state狀態(tài)控制。
      4)AQS由volatile修飾保證多線程可見(jiàn),采用CAS保證原子性。

      分享名稱:AQS源碼分析--jdk1.8
      本文網(wǎng)址:http://www.ef60e0e.cn/article/pjooso.html
      99热在线精品一区二区三区_国产伦精品一区二区三区女破破_亚洲一区二区三区无码_精品国产欧美日韩另类一区
      1. <ul id="0c1fb"></ul>

        <noscript id="0c1fb"><video id="0c1fb"></video></noscript>
        <noscript id="0c1fb"><listing id="0c1fb"><thead id="0c1fb"></thead></listing></noscript>

        石首市| 汉沽区| 苍溪县| 高邑县| 靖江市| 饶阳县| 迭部县| 孝义市| 东安县| 区。| 德令哈市| 永平县| 航空| 安塞县| 岐山县| 眉山市| 中方县| 金溪县| 仲巴县| 安义县| 渑池县| 益阳市| 澎湖县| 游戏| 兴山县| 高邮市| 栾川县| 福泉市| 陈巴尔虎旗| 荔波县| 巨野县| 宁安市| 玛纳斯县| 金秀| 丰县| 娱乐| 来凤县| 锦屏县| 彭山县| 德清县| 和龙市|