java 并发(五)---AbstractQueuedSynchronizer

              1
            public
            class ClhSpinLock implements Lock{
  2privatefinal ThreadLocal<Node> prev;
  3privatefinal ThreadLocal<Node> node;
  4privatefinal AtomicReference<Node> tail = new AtomicReference<Node>(new Node());
  5  6public ClhSpinLock() {
  7this.node = new ThreadLocal<Node>() {
  8protected Node initialValue() {
  9returnnew Node();
 10             }
 11         };
 12 13this.prev = new ThreadLocal<Node>() {
 14protected Node initialValue() {
 15returnnull;
 16             }
 17         };
 18     }
 19 20/**
 21      * 1.初始状态 tail指向一个node(head)节点
 22      * +------+
 23      * | head | <---- tail
 24      * +------+
 25      *
 26      * 2.lock-thread加入等待队列: tail指向新的Node，同时Prev指向tail之前指向的节点
 27      * +----------+
 28      * | Thread-A |
 29      * | := Node  | <---- tail
 30      * | := Prev  | -----> +------+
 31      * +----------+        | head |
 32      *                     +------+
 33      *
 34      *             +----------+            +----------+
 35      *             | Thread-B |            | Thread-A |
 36      * tail ---->  | := Node  |     -->    | := Node  |
 37      *             | := Prev  | ----|      | := Prev  | ----->  +------+
 38      *             +----------+            +----------+         | head |
 39      *                                                          +------+
 40      * 3.寻找当前node的prev-node然后开始自旋
 41      *
 42      */ 43publicvoid lock() {
 44final Node node = this.node.get();
 45         node.locked = true;
 46         Node pred = this.tail.getAndSet(node);
 47this.prev.set(pred);
 48// 自旋 49while (pred.locked);
 50     }
 51 52publicvoid unlock() {
 53final Node node = this.node.get();
 54         node.locked = false;
 55this.node.set(this.prev.get());
 56     }
 57 58     @Override
 59publicvoid lockInterruptibly() throws InterruptedException {
 60// TODO Auto-generated method stub 61 62     }
 63 64     @Override
 65publicboolean tryLock() {
 66// TODO Auto-generated method stub 67returnfalse;
 68     }
 69 70     @Override
 71publicboolean tryLock(long time, TimeUnit unit) throws InterruptedException {
 72// TODO Auto-generated method stub 73returnfalse;
 74     }
 75 76     @Override
 77public Condition newCondition() {
 78// TODO Auto-generated method stub 79returnnull;
 80     }
 81 82privatestaticclass Node {
 83privatevolatileboolean locked;
 84     }
 85 }

              1
            public
            class LockTest {
  2staticint count = 0;
  3  4publicstaticvoid testLock(Lock lock) {
  5try {
  6             lock.lock();
  7for (int i = 0; i < 10000000; i++) ++count;
  8         } finally {
  9             lock.unlock();
 10         }
 11     }
 12 13publicstaticvoid main(String[] args) throws InterruptedException, BrokenBarrierException {
 14final ClhSpinLock clh = new ClhSpinLock();
 15final CyclicBarrier cb = new CyclicBarrier(10, new Runnable() {
 16             @Override
 17publicvoid run() {
 18                 System.out.println(count);
 19             }
 20         });
 21 22for (int i = 0; i < 10; i++) {
 23new Thread(new Runnable() {
 24                 @Override
 25publicvoid run() {
 26                     testLock(clh);
 27// 这段代码是非lock比较使用 28//                    for (int i = 0; i < 10000000; i++) 29//                        count++; 30try {
 31                         cb.await();
 32                     } catch (InterruptedException | BrokenBarrierException e) {
 33                         e.printStackTrace();
 34                     }
 35                 }
 36             }).start();
 37         }
 38     }
 39 }

              1
            static
            final
            class Node {
  2/** Marker to indicate a node is waiting in shared mode */  3// 标识节点当前在共享模式下  4staticfinal Node SHARED = new Node();
  5/** Marker to indicate a node is waiting in exclusive mode */  6// 标识节点当前在独占模式下  7staticfinal Node EXCLUSIVE = null;
  8  9// ======== 下面的几个int常量是给waitStatus用的 =========== 10/** waitStatus value to indicate thread has cancelled */ 11// 代码此线程取消了争抢这个锁 12staticfinalint CANCELLED =  1;
 13/** waitStatus value to indicate successor‘s thread needs unparking */ 14// 官方的描述是，其表示当前node的后继节点对应的线程需要被唤醒 15staticfinalint SIGNAL    = -1;
 16/** waitStatus value to indicate thread is waiting on condition */ 17// 本文不分析condition，所以略过吧，下一篇文章会介绍这个 18staticfinalint CONDITION = -2;
 19/**
 20      * waitStatus value to indicate the next acquireShared should
 21      * unconditionally propagate
 22      */ 23// 同样的不分析，略过吧 24staticfinalint PROPAGATE = -3;
 25// ===================================================== 26 27// 取值为上面的1、-1、-2、-3，或者0(以后会讲到) 28// 这么理解，暂时只需要知道如果这个值 大于0 代表此线程取消了等待， 29// 也许就是说半天抢不到锁，不抢了，ReentrantLock是可以指定timeouot的。。。 30volatileint waitStatus;
 31// 前驱节点的引用 32volatile Node prev;
 33// 后继节点的引用 34volatile Node next;
 35// 这个就是线程本尊 36volatile Thread thread;
 37 38 }

              1
            // 我们看到，这个方法，如果tryAcquire(arg) 返回true, 也就结束了。
              2
            // 否则，acquireQueued方法会将线程压到队列中
              3
            public
            final
            void acquire(int arg) { // 此时 arg == 1  4// 首先调用tryAcquire(1)一下，名字上就知道，这个只是试一试  5// 因为有可能直接就成功了呢，也就不需要进队列排队了，  6// 对于公平锁的语义就是：本来就没人持有锁，根本没必要进队列等待(又是挂起，又是等待被唤醒的)  7if (!tryAcquire(arg) &&
  8// tryAcquire(arg)没有成功，这个时候需要把当前线程挂起，放到阻塞队列中。  9             acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) {
 10               selfInterrupt();
 11         }
 12     }
 13 14 15finalboolean acquireQueued(final Node node, int arg) {
 16boolean failed = true;
 17try {
 18boolean interrupted = false;
 19for (;;) {
 20final Node p = node.predecessor();
 21if (p == head && tryAcquire(arg)) {
 22                 setHead(node);
 23                 p.next = null; // help GC 24                 failed = false;
 25return interrupted;
 26             }
 27if (shouldParkAfterFailedAcquire(p, node) &&
 28                 parkAndCheckInterrupt())

                            // 说明当前线程是被中断唤醒的。
                // 注意：线程被中断之后会继续走到if处去判断，也就是会忽视中断。
                // 除非碰巧线程中断后acquire成功了，那么根据Java的最佳实践，
                // 需要重新设置线程的中断状态（acquire.selfInterrupt）。        
        

             29                 interrupted = true;
 30         }
 31     } finally {
 32if (failed)
 33             cancelAcquire(node);
 34     }
 35 }
 36

              1
            /**
  2      * Creates and enqueues node for current thread and given mode.
  3      *
  4      * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared
  5      * @return the new node
  6      */
              7
            // 此方法的作用是把线程包装成node，同时进入到队列中
              8
            // 参数mode此时是Node.EXCLUSIVE，代表独占模式
              9
            private Node addWaiter(Node mode) {
 10         Node node = new Node(Thread.currentThread(), mode);
 11// Try the fast path of enq; backup to full enq on failure 12// 以下几行代码想把当前node加到链表的最后面去，也就是进到阻塞队列的最后 13         Node pred = tail;
 14 15// tail!=null => 队列不为空(tail==head的时候，其实队列是空的，不过不管这个吧) 16if (pred != null) {
 17// 设置自己的前驱 为当前的队尾节点 18             node.prev = pred;
 19// 用CAS把自己设置为队尾, 如果成功后，tail == node了 20if (compareAndSetTail(pred, node)) {
 21// 进到这里说明设置成功，当前node==tail, 将自己与之前的队尾相连， 22// 上面已经有 node.prev = pred 23// 加上下面这句，也就实现了和之前的尾节点双向连接了 24                 pred.next = node;
 25// 线程入队了，可以返回了 26return node;
 27             }
 28         }
 29// 仔细看看上面的代码，如果会到这里， 30// 说明 pred==null(队列是空的) 或者 CAS失败(有线程在竞争入队) 31// 读者一定要跟上思路，如果没有跟上，建议先不要往下读了，往回仔细看，否则会浪费时间的 32         enq(node);
 33return node;
 34     }

              1
            /**
  2      * Inserts node into queue, initializing if necessary. See picture above.
  3      * @param node the node to insert
  4      * @return node‘s predecessor
  5      */
              6
            // 采用自旋的方式入队
              7
            // 之前说过，到这个方法只有两种可能：等待队列为空，或者有线程竞争入队，
              8
            // 自旋在这边的语义是：CAS设置tail过程中，竞争一次竞争不到，我就多次竞争，总会排到的
              9
            private Node enq(final Node node) {
 10for (;;) {
 11             Node t = tail;
 12// 之前说过，队列为空也会进来这里 13if (t == null) { // Must initialize 14// 初始化head节点 15// 细心的读者会知道原来head和tail初始化的时候都是null，反正我不细心 16// 还是一步CAS，你懂的，现在可能是很多线程同时进来呢 17if (compareAndSetHead(new Node()))
 18// 给后面用：这个时候head节点的waitStatus==0, 看new Node()构造方法就知道了 19 20// 这个时候有了head，但是tail还是null，设置一下， 21// 把tail指向head，放心，马上就有线程要来了，到时候tail就要被抢了 22// 注意：这里只是设置了tail=head，这里可没return哦，没有return，没有return 23// 所以，设置完了以后，继续for循环，下次就到下面的else分支了 24                     tail = head;
 25             } else {
 26// 下面几行，和上一个方法 addWaiter 是一样的， 27// 只是这个套在无限循环里，反正就是将当前线程排到队尾，有线程竞争的话排不上重复排 28                 node.prev = t;
 29if (compareAndSetTail(t, node)) {
 30                     t.next = node;
 31return t;
 32                 }
 33             }
 34         }
 35     }
 36

              1
            /**
  2      * Checks and updates status for a node that failed to acquire.
  3      * Returns true if thread should block. This is the main signal
  4      * control in all acquire loops.  Requires that pred == node.prev
  5      *
  6      * @param pred node‘s predecessor holding status
  7      * @param node the node
  8      * @return {@code true} if thread should block
  9      */
             10
            // 刚刚说过，会到这里就是没有抢到锁呗，这个方法说的是："当前线程没有抢到锁，是否需要挂起当前线程？"
             11
            // 第一个参数是前驱节点，第二个参数才是代表当前线程的节点
             12
            private
            static
            boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
 13int ws = pred.waitStatus;
 14// 前驱节点的 waitStatus == -1 ，说明前驱节点状态正常，当前线程需要挂起，直接可以返回true 15if (ws == Node.SIGNAL)
 16/*
 17              * This node has already set status asking a release
 18              * to signal it, so it can safely park.
 19              */ 20returntrue;
 21 22// 前驱节点 waitStatus大于0 ，之前说过，大于0 说明前驱节点取消了排队。这里需要知道这点： 23// 进入阻塞队列排队的线程会被挂起，而唤醒的操作是由前驱节点完成的。 24// 所以下面这块代码说的是将当前节点的prev指向waitStatus<=0的节点， 25// 简单说，就是为了找个好爹，因为你还得依赖它来唤醒呢，如果前驱节点取消了排队， 26// 找前驱节点的前驱节点做爹，往前循环总能找到一个好爹的 27if (ws > 0) {
 28/*
 29              * Predecessor was cancelled. Skip over predecessors and
 30              * indicate retry.
 31              */ 32do {
 33                 node.prev = pred = pred.prev;
 34             } while (pred.waitStatus > 0);
 35             pred.next = node;
 36         } else {
 37/*
 38              * waitStatus must be 0 or PROPAGATE.  Indicate that we
 39              * need a signal, but don‘t park yet.  Caller will need to
 40              * retry to make sure it cannot acquire before parking.
 41              */ 42// 仔细想想，如果进入到这个分支意味着什么 43// 前驱节点的waitStatus不等于-1和1，那也就是只可能是0，-2，-3 44// 在我们前面的源码中，都没有看到有设置waitStatus的，所以每个新的node入队时，waitStatu都是0 45// 用CAS将前驱节点的waitStatus设置为Node.SIGNAL(也就是-1) 46             compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
 47         }
 48returnfalse;
 49     }
 50 51// private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) 52// 这个方法结束根据返回值我们简单分析下： 53// 如果返回true, 说明前驱节点的waitStatus==-1，是正常情况，那么当前线程需要被挂起，等待以后被唤醒 54//        我们也说过，以后是被前驱节点唤醒，就等着前驱节点拿到锁，然后释放锁的时候叫你好了 55// 如果返回false, 说明当前不需要被挂起，为什么呢？往后看 56 57// 跳回到前面是这个方法 58// if (shouldParkAfterFailedAcquire(p, node) && 59//                parkAndCheckInterrupt()) 60//                interrupted = true; 61 62// 1. 如果shouldParkAfterFailedAcquire(p, node)返回true， 63// 那么需要执行parkAndCheckInterrupt(): 64 65// 这个方法很简单，因为前面返回true，所以需要挂起线程，这个方法就是负责挂起线程的 66// 这里用了LockSupport.park(this)来挂起线程，然后就停在这里了，等待被唤醒======= 67privatefinalboolean parkAndCheckInterrupt() {
 68         LockSupport.park(this);
 69return Thread.interrupted();
 70     }
 71 72// 2. 接下来说说如果shouldParkAfterFailedAcquire(p, node)返回false的情况 73 74// 仔细看shouldParkAfterFailedAcquire(p, node)，我们可以发现，其实第一次进来的时候，一般都不会返回true的，原因很简单，前驱节点的waitStatus=-1是依赖于后继节点设置的。也就是说，我都还没给前驱设置-1呢，怎么可能是true呢，但是要看到，这个方法是套在循环里的，所以第二次进来的时候状态就是-1了。 75 76// 解释下为什么shouldParkAfterFailedAcquire(p, node)返回false的时候不直接挂起线程： 77// => 是为了应对在经过这个方法后，node已经是head的直接后继节点了。剩下的读者自己想想吧。 78 }

              1
            public
            final
            boolean release(int arg) {
  2// tryReease由子类实现，通过设置state值来达到同步的效果。  3if (tryRelease(arg)) {
  4         Node h = head;
  5// waitStatus为0说明是初始化的空队列  6if (h != null && h.waitStatus != 0)
  7// 唤醒后续的结点  8             unparkSuccessor(h);
  9returntrue;
 10     }
 11returnfalse;
 12 }

              1
            public
            final
            void acquireShared(int arg) {
  2//如果没有许可了则入队等待  3if (tryAcquireShared(arg) < 0)
  4         doAcquireShared(arg);
  5 }
  6  7privatevoid doAcquireShared(int arg) {
  8// 添加队列  9final Node node = addWaiter(Node.SHARED);
 10boolean failed = true;
 11try {
 12boolean interrupted = false;
 13// 等待前继释放并传递 14for (;;) {
 15final Node p = node.predecessor();
 16if (p == head) {
 17int r = tryAcquireShared(arg);// 尝试获取 18if (r >= 0) {
 19// 获取成功则前继出队，跟独占不同的是 20// 会往后面结点传播唤醒的操作，保证剩下等待的线程能够尽快 获取到剩下的许可。 21                     setHeadAndPropagate(node, r);
 22                     p.next = null; // help GC 23if (interrupted)
 24                         selfInterrupt();
 25                     failed = false;
 26return;
 27                 }
 28             }
 29 30// p != head || r < 0 31if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
 32                 interrupted = true;
 33         }
 34     }
 35finally {
 36if (failed)
 37             cancelAcquire(node);
 38     }
 39 }

              1
            private
            void setHeadAndPropagate(Node node, int propagate) {
  2     Node h = head; // Record old head for check below  3     setHead(node);
  4/*
  5      * 尝试唤醒后继的结点：<br />
  6      * propagate > 0说明许可还有能够继续被线程acquire;<br />
  7      * 或者 之前的head被设置为PROPAGATE(PROPAGATE可以被转换为SIGNAL)说明需要往后传递;<br />
  8      * 或者为null,我们还不确定什么情况。 <br />
  9      * 并且 后继结点是共享模式或者为如上为null。
 10      * <p>
 11      * 上面的检查有点保守，在有多个线程竞争获取/释放的时候可能会导致不必要的唤醒。<br />
 12      *
 13      */ 14if (propagate > 0 || h == null || h.waitStatus < 0) {
 15         Node s = node.next;
 16// 后继结是共享模式或者s == null（不知道什么情况） 17// 如果后继是独占模式，那么即使剩下的许可大于0也不会继续往后传递唤醒操作 18// 即使后面有结点是共享模式。 19if (s == null || s.isShared())
 20// 唤醒后继结点 21             doReleaseShared();
 22     }
 23 }
 24 25privatevoid doReleaseShared() {
 26for (;;) {
 27         Node h = head;
 28// 队列不为空且有后继结点 29if (h != null && h != tail) {
 30int ws = h.waitStatus;
 31// 不管是共享还是独占只有结点状态为SIGNAL才尝试唤醒后继结点 32if (ws == Node.SIGNAL) {
 33// 将waitStatus设置为0 34if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
 35continue; // loop to recheck cases 36                 unparkSuccessor(h);// 唤醒后继结点 37// 如果状态为0则更新状态为PROPAGATE，更新失败则重试 38             } elseif (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
 39continue; // loop on failed CAS 40         }
 41// 如果过程中head被修改了则重试。 42if (h == head) // loop if head changed 43break;
 44     }
 45 }

              1
            public
            final
            boolean releaseShared(int arg) {
  2if (tryReleaseShared(arg)) {
  3         doReleaseShared();
  4returntrue;
  5     }
  6returnfalse;
  7 }