多线程并发AQS源码分析

古城微笑少年丶 2022-09-12 14:53 298阅读 0赞

一、业务场景描述

业务场景，银行中存在一个办理业务的窗口，A、B、C三个用户（线程）去办理业务，先是A用户去窗口办理业务，然后B、C在等候区排队等候，过一会儿，A用户（线程）办理完业务，B用户（线程）去业务窗口办理业务，以此类推C用户一样。

二、分析之前上分析图

watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6ZOB5rGJ5p-U5oOFbGk_size_20_color_FFFFFF_t_70_g_se_x_16

watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6ZOB5rGJ5p-U5oOFbGk_size_20_color_FFFFFF_t_70_g_se_x_16 1

三、源码分析

  1. #(1)A线程开始工作，第一步调用lock方法
  2. public void lock() {
  3.         sync.lock();
  4.     }
  5. #其中：sync继承 AbstractQueuedSynchronizer——（AQS）
  6.   abstract static class Sync extends AbstractQueuedSynchronizer
  7. #(2)进入lock方法体
  8.    final void lock() {
  9. #比较并设置state的状态为1；1-表示占用，0-表示未占用
  10.           if (compareAndSetState(0, 1))
  11.               setExclusiveOwnerThread(Thread.currentThread());
  12.           else
  13.               acquire(1);
  14.       }
  15. #（3）进入setExclusiveOwnerThread（）设置当前拥有独占访问权限的线程。
  16. 到此A线程开始在窗口执行工作，此代码块到此执行完成。
  18. #（4）用户（线程）B开启
  19.     进入方法
  20.     final void lock() {
  21.     if (compareAndSetState(0, 1))
  22.         setExclusiveOwnerThread(Thread.currentThread());
  23.     else
  24.         #进入这个代码块
  25.         acquire(1);
  26.     }
  27. #（5）acquire的方法结构
  28.     public final void acquire(int arg) {
  29.         if (!tryAcquire(arg) &&
  30.             acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
  31.             selfInterrupt();
  32.     }
  33.     protected final boolean tryAcquire(int acquires) {
  34.             return nonfairTryAcquire(acquires);
  35.         }
  36. #（6）先进入tryAcquire（）方法解析
  37.     protected final boolean tryAcquire(int acquires) {
  38.         return nonfairTryAcquire(acquires);
  39.     }
  41. #（7）进入nonfairTryAcquire方法 传入参数为1
  43.     final boolean nonfairTryAcquire(int acquires) {
  44.         final Thread current = Thread.currentThread();
  45.         #c=2不进入
  46.         int c = getState();
  47.         if (c == 0) {
  48.            if (compareAndSetState(0, acquires)) {
  49.             setExclusiveOwnerThread(current);
  50.             return true;
  51.              }
  52.            }
  53.         #current为B线程，getExclusiveOwnerThread()为A线程
  54.         #所以也不进入
  55.             else if (current == getExclusiveOwnerThread()) {
  56.                 int nextc = c + acquires;
  57.                 if (nextc < 0) // overflow
  58.                     throw new Error("Maximum lock count exceeded");
  59.                 setState(nextc);
  60.                 return true;
  61.             }
  62.             #返回false
  63.             return false;
  64.         }
  65. #（8）向下执行addWaiter（）方法
  67.     public final void acquire(int arg) {
  68.     if (!tryAcquire(arg) &&
  69.         acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
  70.         selfInterrupt();
  71.     }
  73. #（9）进入方法体addWaiter（）
  74.     private Node addWaiter(Node mode) {
  75.     Node node = new Node(Thread.currentThread(), mode);
  76.     // Try the fast path of enq; backup to full enq on failure
  77.     Node pred = tail;
  78.     #第一次进入pred 为空所以不进入
  79.     if (pred != null) {
  80.         node.prev = pred;
  81.         if (compareAndSetTail(pred, node)) {
  82.             pred.next = node;
  83.             return node;
  84.         }
  85.     }
  86.     #进入下面enq
  87.     enq(node);
  88.     return node;
  89.     }
  90. #（10）进入enq方法
  91.     private Node enq(final Node node) {
  92.         for (;;) {
  93.             Node t = tail;
  94.             #满足条件进入
  95.             if (t == null) { // Must initialize
  96.                 #创建一个哨兵节点，头节点的地址指向一个空node(哨兵节点)
  97.                 if (compareAndSetHead(new Node()))
  98.                     #刚刚创建的头节点指向尾节点
  99.                     tail = head;
  100.             } else {
  101.                 node.prev = t;
  102.                 if (compareAndSetTail(t, node)) {
  103.                     t.next = node;
  104.                     return t;
  105.                 }
  106.             }
  107.         }
  108.     }
  110.     #进入二次循环
  111.    private Node enq(final Node node) {    
  112.         for (;;) {
  113.             Node t = tail;
  114.             #条件不满足，因为上次循环已经初始化
  115.             if (t == null) { // Must initialize
  116.                 if (compareAndSetHead(new Node()))
  117.                     #刚刚创建的头节点指向尾节点
  118.                     tail = head;
  119.             #满足else条件进入下面代码块
  120.             } else {
  121.                 #哨兵节点的前指针指向初始节点
  122.                 node.prev = t;
  123.                 #进入下面代码compareAndSetTail，将尾指针指向新的B节点
  124.                 #
  125.                 if (compareAndSetTail(t, node)) {
  126.                     #将哨兵节点的后指针指向node(B节点)
  127.                     t.next = node;
  128.                     #返回哨兵节点，然后此方法结束
  129.                     return t;
  130.                 }
  131.             }
  132.         }
  133.     }
  134. #（11）addWaiter方法结束，放回B的node节点
  136. #（12）进入acquireQueued方法，注意当前还处于B线程中
  137.     public final void acquire(int arg) {
  138.         if (!tryAcquire(arg) &&
  139.             #arg=1,addWaiter(Node.EXCLUSIVE)为B节点
  140.             acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
  141.             selfInterrupt();
  142.     }
  143. #（13）进入acquireQueued方法
  145.     final boolean acquireQueued(final Node node, int arg) {
  146.         boolean failed = true;
  147.         try {
  148.             boolean interrupted = false;
  149.             for (;;) {
  150.                 #返回前置节点，也就是哨兵节点
  151.                 final Node p = node.predecessor();
  152.                 #p == head 结果为true
  153.                 if (p == head && tryAcquire(arg)) {
  154.                     setHead(node);
  155.                     p.next = null; // help GC
  156.                     failed = false;
  157.                     return interrupted;
  158.                 }
  159.                 if (shouldParkAfterFailedAcquire(p, node) &&
  160.                     parkAndCheckInterrupt())
  161.                     interrupted = true;
  162.             }
  163.         } finally {
  164.             if (failed)
  165.                 cancelAcquire(node);
  166.         }
  167.     }
  168. #（14）分析一下tryAcquire(arg)方法
  169.     protected final boolean tryAcquire(int acquires) {
  170.                 return nonfairTryAcquire(acquires);
  171.     }
  173.     #进入nonfairTryAcquire(acquires);方法
  174.     final boolean nonfairTryAcquire(int acquires) {
  175.                 final Thread current = Thread.currentThread();
  176.                 #state = 1;
  177.                 int c = getState();
  178.                 #故不满足条件，不进入
  179.                 if (c == 0) {
  180.                     if (compareAndSetState(0, acquires)) {
  181.                         setExclusiveOwnerThread(current);
  182.                         return true;
  183.                     }
  184.                 }
  185.                 #current为B线程，getExclusiveOwnerThread（）为A线程，故不进入这个代码块
  186.                 else if (current == getExclusiveOwnerThread()) {
  187.                     int nextc = c + acquires;
  188.                     if (nextc < 0) // overflow
  189.                         throw new Error("Maximum lock count exceeded");
  190.                     setState(nextc);
  191.                     return true;
  192.                 }
  193.                 #返回false
  194.                 return false;
  195.             }
  196. #（15）拿到tryAcquire的结果后继续分析acquireQueued方法
  197.     final boolean acquireQueued(final Node node, int arg) {
  198.             boolean failed = true;
  199.             try {
  200.                 boolean interrupted = false;
  201.                 for (;;) {
  202.                     final Node p = node.predecessor();
  203.                     #条件不满足，p == head结果为true,tryAcquire(arg)结果为false,所以不进入if
  204.                     if (p == head && tryAcquire(arg)) {
  205.                         setHead(node);
  206.                         p.next = null; // help GC
  207.                         failed = false;
  208.                         return interrupted;
  209.                     }
  210.                     #进入shouldParkAfterFailedAcquire(p, node) 方法
  211.                     if (shouldParkAfterFailedAcquire(p, node) &&
  212.                         parkAndCheckInterrupt())
  213.                         interrupted = true;
  214.                 }
  215.             } finally {
  216.                 if (failed)
  217.                     cancelAcquire(node);
  218.             }
  219.         }
  221. #（16）分析shouldParkAfterFailedAcquire(p, node)方法，p为哨兵节点，node为B节点
  223.     private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
  224.             #ws为0
  225.             int ws = pred.waitStatus;
  226.             if (ws == Node.SIGNAL)
  227.                 /*
  228.                  * This node has already set status asking a release
  229.                  * to signal it, so it can safely park.
  230.                  */
  231.                 return true;
  232.             if (ws > 0) {
  233.                 /*
  234.                  * Predecessor was cancelled. Skip over predecessors and
  235.                  * indicate retry.
  236.                  */
  237.                 do {
  238.                     node.prev = pred = pred.prev;
  239.                 } while (pred.waitStatus > 0);
  240.                 pred.next = node;
  242.             #进入else
  243.             } else {
  244.                 /*
  245.                  * waitStatus must be 0 or PROPAGATE.  Indicate that we
  246.                  * need a signal, but don't park yet.  Caller will need to
  247.                  * retry to make sure it cannot acquire before parking.
  248.                  */
  249.                 #将哨兵节点的waitStatus设置成-1
  250.                 compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
  251.             }
  252.             #最后返回false
  253.             return false;
  254.         }
  256. #（17）拿到shouldParkAfterFailedAcquire(p, node)的返回结果为false，继续分析acquireQueued(final Node node, int arg)方法
  257.     final boolean acquireQueued(final Node node, int arg) {
  258.             boolean failed = true;
  259.             try {
  260.                 boolean interrupted = false;
  261.                 for (;;) {
  262.                     final Node p = node.predecessor();
  263.                     if (p == head && tryAcquire(arg)) {
  264.                         setHead(node);
  265.                         p.next = null; // help GC
  266.                         failed = false;
  267.                         return interrupted;
  268.                     }
  269.                     #shouldParkAfterFailedAcquire(p, node) 结果为false不进入
  270.                     if (shouldParkAfterFailedAcquire(p, node) &&
  271.                         parkAndCheckInterrupt())
  272.                         interrupted = true;
  273.                 }
  274.             } finally {
  275.                 if (failed)
  276.                     cancelAcquire(node);
  277.             }
  278.         }
  280. #（18）再次执行acquireQueued的第二次循环
  281.     final boolean acquireQueued(final Node node, int arg) {
  282.                 boolean failed = true;
  283.                 try {
  284.                     boolean interrupted = false;
  285.                     for (;;) {
  286.                         final Node p = node.predecessor();
  287.                         #p == head满足，同样tryAcquire(arg)返回false不满足，不进入
  288.                         if (p == head && tryAcquire(arg)) {
  289.                             setHead(node);
  290.                             p.next = null; // help GC
  291.                             failed = false;
  292.                             return interrupted;
  293.                         }
  294.                         #进入这个节点继续分析
  295.                         if (shouldParkAfterFailedAcquire(p, node) &&
  296.                             parkAndCheckInterrupt())
  297.                             interrupted = true;
  298.                     }
  299.                 } finally {
  300.                     if (failed)
  301.                         cancelAcquire(node);
  302.                 }
  303.             }
  304. #（19）再次分析shouldParkAfterFailedAcquire(p, node)方法
  306.     private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
  307.             #等于-1
  308.             int ws = pred.waitStatus;
  309.             #符合条件进入
  310.             if (ws == Node.SIGNAL)
  311.                 /*
  312.                  * This node has already set status asking a release
  313.                  * to signal it, so it can safely park.
  314.                  */
  315.                 #返回true
  316.                 return true;
  317.             if (ws > 0) {
  318.                 /*
  319.                  * Predecessor was cancelled. Skip over predecessors and
  320.                  * indicate retry.
  321.                  */
  322.                 do {
  323.                     node.prev = pred = pred.prev;
  324.                 } while (pred.waitStatus > 0);
  325.                 pred.next = node;
  326.             } else {
  327.                 /*
  328.                  * waitStatus must be 0 or PROPAGATE.  Indicate that we
  329.                  * need a signal, but don't park yet.  Caller will need to
  330.                  * retry to make sure it cannot acquire before parking.
  331.                  */
  332.                 compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
  333.             }
  334.             return false;
  335.         }
  337. #（20）根据shouldParkAfterFailedAcquire返回的结果（true），继续执行下一步parkAndCheckInterrupt()
  339.     final boolean acquireQueued(final Node node, int arg) {
  340.             boolean failed = true;
  341.             try {
  342.                 boolean interrupted = false;
  343.                 for (;;) {
  344.                     final Node p = node.predecessor();
  345.                     if (p == head && tryAcquire(arg)) {
  346.                         setHead(node);
  347.                         p.next = null; // help GC
  348.                         failed = false;
  349.                         return interrupted;
  350.                     }
  351.                     if (shouldParkAfterFailedAcquire(p, node) &&
  352.                         parkAndCheckInterrupt())
  353.                         interrupted = true;
  354.                 }
  355.             } finally {
  356.                 if (failed)
  357.                     cancelAcquire(node);
  358.             }
  359.         }
  361.     #当前的B线程开始阻塞，因为执行了LockSupport.park(this)；待A线程结束的时候，唤醒B
  362.     private final boolean parkAndCheckInterrupt() {
  363.             LockSupport.park(this);
  364.             return Thread.interrupted();
  365.         }
  367. #（21）C线程也是像B一样，最后阻塞等待被唤醒
  369.     private final boolean parkAndCheckInterrupt() {
  370.             LockSupport.park(this);
  372.             #如果当前线程已被中断，则为true ； 否则为false
  373.             return Thread.interrupted();
  374.         }
  377. #（22）此时A线业务办理完成，调用lock.unlock();方法解锁
  378.     public void unlock() {
  379.             sync.release(1);
  380.         }
  382.     #进入release方法,arg = 1
  383.       public final boolean release(int arg) {
  384.           if (tryRelease(arg)) {
  385.               Node h = head;
  386.               if (h != null && h.waitStatus != 0)
  387.                   unparkSuccessor(h);
  388.               return true;
  389.           }
  390.           return false;
  391.       }
  392. #（23）进入tryRelease(int releases)方法，releases = 1；
  393.     protected final boolean tryRelease(int releases) {
  394.         # c = 0;
  395.         int c = getState() - releases;
  396.         #相等不进入
  397.         if (Thread.currentThread() != getExclusiveOwnerThread())
  398.             throw new IllegalMonitorStateException();
  399.         boolean free = false;
  400.         #进入
  401.         if (c == 0) {
  402.             free = true;
  403.             #将原来的A线程变为null
  404.             setExclusiveOwnerThread(null);
  405.         }
  406.         #设置当前状态为0;state = 1
  407.         setState(c);
  408.         #返回free = true;
  409.         return free;
  410.     }
  412. #（24）返回release(int arg)方法，继续向下执行
  413.       public final boolean release(int arg) {
  414.             #返回true进入
  415.           if (tryRelease(arg)) {
  416.               Node h = head;
  417.                 #此时h为哨兵节点，h.waitStatus 为 -1；所以进入
  418.               if (h != null && h.waitStatus != 0)
  419.                     #接下来看该方法
  420.                   unparkSuccessor(h);
  421.               return true;
  422.           }
  423.           return false;
  424.       }
  426. #（25）解析unparkSuccessor(Node node)方法，其中node为哨兵节点
  427.     private void unparkSuccessor(Node node) {
  428.         /*
  429.          * If status is negative (i.e., possibly needing signal) try
  430.          * to clear in anticipation of signalling.  It is OK if this
  431.          * fails or if status is changed by waiting thread.
  432.          */
  433.         int ws = node.waitStatus;
  434.         if (ws < 0)
  435.             #进入，将哨兵node节点的waitStatus，由-1改为0
  436.             compareAndSetWaitStatus(node, ws, 0);
  439.         /*
  440.          * Thread to unpark is held in successor, which is normally
  441.          * just the next node.  But if cancelled or apparently null,
  442.          * traverse backwards from tail to find the actual
  443.          * non-cancelled successor.
  444.          */
  445.         #s为B结点
  446.         Node s = node.next;
  447.         #s!=null，所以不进入
  448.         if (s == null || s.waitStatus > 0) {
  449.             s = null;
  450.             for (Node t = tail; t != null && t != node; t = t.prev)
  451.                 if (t.waitStatus <= 0)
  452.                     s = t;
  453.         }
  454.         #进入
  455.         if (s != null)
  456.             #唤醒B进程
  457.             LockSupport.unpark(s.thread);
  458.     }
  459. #（26）到此lock.unlock（）方法执行完毕，执行结果：A线程执行完成，B线程开始执行；
  461. #（27）B线程被唤醒后，继续执行后面的代码
  462.     private final boolean parkAndCheckInterrupt() {
  463.         LockSupport.park(this);
  464.         return Thread.interrupted();
  465.     }
  467. #（28）继续执行for循环，node为B节点，arg为1
  468.     final boolean acquireQueued(final Node node, int arg) {
  469.         boolean failed = true;
  470.         try {
  471.             boolean interrupted = false;
  472.             for (;;) {
  473.                 final Node p = node.predecessor();
  474.                 #再次进入tryAcquire（arg）
  475.                 if (p == head && tryAcquire(arg)) {
  476.                     setHead(node);
  477.                     p.next = null; // help GC
  478.                     failed = false;
  479.                     return interrupted;
  480.                 }
  481.                 if (shouldParkAfterFailedAcquire(p, node) &&
  482.                     parkAndCheckInterrupt())
  483.                     interrupted = true;
  484.             }
  485.         } finally {
  486.             if (failed)
  487.                 cancelAcquire(node);
  488.         }
  489.     }
  490.     #的内部方法
  491.       protected final boolean tryAcquire(int acquires) {
  492.           return nonfairTryAcquire(acquires);
  493.       }
  495. #（29）再次分析nonfairTryAcquire(acquires)方法；acquires = 1；
  497.     final boolean nonfairTryAcquire(int acquires) {
  498.         #B线程
  499.         final Thread current = Thread.currentThread();
  500.         # c = 0；
  501.         int c = getState();
  502.         #进入条件
  503.         if (c == 0) {
  504.             #修改状态为1
  505.             if (compareAndSetState(0, acquires)) {
  506.                 #设置当前线程B为有独占访问权限的线程
  507.                 setExclusiveOwnerThread(current);
  508.                 #设置返回值为true
  509.                 return true;
  510.             }
  511.         }
  512.         else if (current == getExclusiveOwnerThread()) {
  513.             int nextc = c + acquires;
  514.             if (nextc < 0) // overflow
  515.                 throw new Error("Maximum lock count exceeded");
  516.             setState(nextc);
  517.             return true;
  518.         }
  519.         return false;
  520.     }
  522.     #tryAcquire(int acquires) 返回true
  523.         protected final boolean tryAcquire(int acquires) {
  524.         return nonfairTryAcquire(acquires);
  525.     }
  527. #（30）返回上层方法acquireQueued(final Node node, int arg)，node为B节点，arg为1
  529.     final boolean acquireQueued(final Node node, int arg) {
  530.         boolean failed = true;
  531.         try {
  532.             boolean interrupted = false;
  533.             for (;;) {
  534.                 final Node p = node.predecessor();
  535.                 #p==head = true; tryAcquire(arg))=true
  536.                 if (p == head && tryAcquire(arg)) {
  537.                     #设置B节点为头节点
  538.                     setHead(node);
  539.                     #哨兵节点的后指针为null,此时的哨兵node,为不可达状态，gc时会被回收掉
  540.                     p.next = null; // help GC
  541.                     failed = false;
  542.                     return interrupted;
  543.                 }
  544.                 if (shouldParkAfterFailedAcquire(p, node) &&
  545.                     parkAndCheckInterrupt())
  546.                     interrupted = true;
  547.             }
  548.         } finally {
  549.             if (failed)
  550.                 cancelAcquire(node);
  551.         }
  552.     }
  554.         #node为B节点
  555.         private void setHead(Node node) {
  556.             #B节点为head节点
  557.             head = node;
  558.             node.thread = null;
  559.             #B节点取消前置节点
  560.             node.prev = null;
  561.         }
  562. #（31）到此lock.lock（）方法执行结束。

发表评论

表情：
评论列表 （有 0 条评论，298人围观）

还没有评论，来说两句吧...

相关阅读

    相关 源码分析:AQS源码

    在开始这篇源码之前，最好先看下转载整理的[这篇文章][Link 1]，有很多值得学习的地方。AQS是用来构建锁或者其他同步组件的基础框架。总体来说，它使用一个 int 成员变量

    冷不防冷不防/ 0 赞/ 601 阅读

    相关 并发-AQS源码分析

    微信搜索：“二十同学” 公众号，欢迎关注一条不一样的成长之路 一、概述 　　谈到并发，不得不谈ReentrantLock；而谈到ReentrantLock，不得不谈

    忘是亡心i忘是亡心i/ 0 赞/ 557 阅读