[关闭]
@huangyichun 2017-08-30T16:07:54.000000Z 字数 8218 阅读 933

AbstractQueuedSynchronizer队列同步器

多线程

更改同步状态state

  1.         //同步器状态
  2.         private volatile int state;
  4.         //设置当前同步状态
  5.         protected final void setState(int newState) {
  6.             state = newState;
  7.         }
  9.         //采用CAS算法更新state状态
  10.         protected final boolean compareAndSetState(int expect, int update) {
  11.             // See below for intrinsics setup to support this
  12.             return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
  13.         }

同步队列FIFO双向队列

  1.     static final class Node {
  2.             //标记节点的等待模式为共享型模式
  3.             static final Node SHARED = new Node();
  4.             //标记节点的等待模式为独占型模式
  5.             static final Node EXCLUSIVE = null;
  7.             /** waitStatus value to indicate thread has cancelled */
  8.             static final int CANCELLED =  1;
  9.             /** waitStatus value to indicate successor's thread needs unparking */
  10.             static final int SIGNAL    = -1;
  11.             /** waitStatus value to indicate thread is waiting on condition */
  12.             static final int CONDITION = -2;
  13.             /**
  14.              * waitStatus value to indicate the next acquireShared should
  15.              * unconditionally propagate
  16.              */
  17.             static final int PROPAGATE = -3;
  19.             /**
  20.              * Status field, taking on only the values:
  21.              *   SIGNAL:     The successor of this node is (or will soon be)
  22.              *               blocked (via park), so the current node must
  23.              *               unpark its successor when it releases or
  24.              *               cancels. To avoid races, acquire methods must
  25.              *               first indicate they need a signal,
  26.              *               then retry the atomic acquire, and then,
  27.              *               on failure, block.
  28.              *   CANCELLED:  This node is cancelled due to timeout or interrupt.
  29.              *               Nodes never leave this state. In particular,
  30.              *               a thread with cancelled node never again blocks.
  31.              *   CONDITION:  This node is currently on a condition queue.
  32.              *               It will not be used as a sync queue node
  33.              *               until transferred, at which time the status
  34.              *               will be set to 0. (Use of this value here has
  35.              *               nothing to do with the other uses of the
  36.              *               field, but simplifies mechanics.)
  37.              *   PROPAGATE:  A releaseShared should be propagated to other
  38.              *               nodes. This is set (for head node only) in
  39.              *               doReleaseShared to ensure propagation
  40.              *               continues, even if other operations have
  41.              *               since intervened.
  42.              *   0:          None of the above
  43.              *
  44.              * The values are arranged numerically to simplify use.
  45.              * Non-negative values mean that a node doesn't need to
  46.              * signal. So, most code doesn't need to check for particular
  47.              * values, just for sign.
  48.              *
  49.              * The field is initialized to 0 for normal sync nodes, and
  50.              * CONDITION for condition nodes.  It is modified using CAS
  51.              * (or when possible, unconditional volatile writes).
  52.              */
  53.             volatile int waitStatus;
  55.             /**
  56.              * Link to predecessor node that current node/thread relies on
  57.              * for checking waitStatus. Assigned during enqueuing, and nulled
  58.              * out (for sake of GC) only upon dequeuing.  Also, upon
  59.              * cancellation of a predecessor, we short-circuit while
  60.              * finding a non-cancelled one, which will always exist
  61.              * because the head node is never cancelled: A node becomes
  62.              * head only as a result of successful acquire. A
  63.              * cancelled thread never succeeds in acquiring, and a thread only
  64.              * cancels itself, not any other node.
  65.              */
  66.             volatile Node prev;
  68.             /**
  69.              * Link to the successor node that the current node/thread
  70.              * unparks upon release. Assigned during enqueuing, adjusted
  71.              * when bypassing cancelled predecessors, and nulled out (for
  72.              * sake of GC) when dequeued.  The enq operation does not
  73.              * assign next field of a predecessor until after attachment,
  74.              * so seeing a null next field does not necessarily mean that
  75.              * node is at end of queue. However, if a next field appears
  76.              * to be null, we can scan prev's from the tail to
  77.              * double-check.  The next field of cancelled nodes is set to
  78.              * point to the node itself instead of null, to make life
  79.              * easier for isOnSyncQueue.
  80.              */
  81.             volatile Node next;
  83.             /**
  84.              * 获取同步状态的线程
  85.              */
  86.             volatile Thread thread;
  88.             /**
  89.             *等待队列中的后继节点。如果当前节点是共享的，那么这个字段将是一个SHARED常量
  90.             *也就是说节点类型和等待队列中的后继节点共用一个字段
  91.             */
  92.             Node nextWaiter; 
  94.             /**
  95.              * Returns true if node is waiting in shared mode.
  96.              */
  97.             final boolean isShared() {
  98.                 return nextWaiter == SHARED;
  99.             }
  101.             /**
  102.              * Returns previous node, or throws NullPointerException if null.
  103.              * Use when predecessor cannot be null.  The null check could
  104.              * be elided, but is present to help the VM.
  105.              *
  106.              * @return the predecessor of this node
  107.              */
  108.             final Node predecessor() throws NullPointerException {
  109.                 Node p = prev;
  110.                 if (p == null)
  111.                     throw new NullPointerException();
  112.                 else
  113.                     return p;
  114.             }
  116.             Node() {    // Used to establish initial head or SHARED marker
  117.             }
  119.             Node(Thread thread, Node mode) {     // Used by addWaiter
  120.                 this.nextWaiter = mode;
  121.                 this.thread = thread;
  122.             }
  124.             Node(Thread thread, int waitStatus) { // Used by Condition
  125.                 this.waitStatus = waitStatus;
  126.                 this.thread = thread;
  127.             }
  128.         }
  1.      private void setHead(Node node) {
  2.             head = node;
  3.             node.thread = null;
  4.             node.prev = null;
  5.         }

独占式同步状态的获取与释放

  1.      public final void acquire(int arg) {
  2.             if (!tryAcquire(arg) &&
  3.                 acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
  4.                 selfInterrupt();
  5.         }
  7.     //保证线程安全的获取同步状态，该方法需要重写
  8.     protected boolean tryAcquire(int arg) {
  9.         throw new UnsupportedOperationException();
  10.     }
  12.     //添加同步节点
  13.     private Node addWaiter(Node mode) {
  14.         Node node = new Node(Thread.currentThread(), mode);//构造同步节点
  15.         // Try the fast path of enq; backup to full enq on failure
  16.         Node pred = tail;
  17.         if (pred != null) {//如果队列不为空
  18.             node.prev = pred;
  19.             if (compareAndSetTail(pred, node)) {//在尾部添加节点
  20.                 pred.next = node;
  21.                 return node; //返回节点
  22.             }
  23.         }
  24.         enq(node); //队列为空
  25.         return node;
  26.     }
  28.      private Node enq(final Node node) {
  29.         for (;;) {//无限循环
  30.             Node t = tail;
  31.             if (t == null) { // Must initialize
  32.                 if (compareAndSetHead(new Node()))//添加一个空的节点，作为头节点
  33.                     tail = head;
  34.             } else {
  35.                 node.prev = t;  
  36.                 if (compareAndSetTail(t, node)) {//将节点插入到末尾,CAS算法保障
  37.                     t.next = node;
  38.                     return t;
  39.                 }
  40.             }
  41.         }
  42.     }
  45.      final boolean acquireQueued(final Node node, int arg) {
  46.         boolean failed = true;
  47.         try {
  48.             boolean interrupted = false;
  49.             for (;;) {
  50.                 final Node p = node.predecessor();//获取前一个节点
  51.                 if (p == head && tryAcquire(arg)) {//前一个节点为头节点，且获取同步状态
  52.                     setHead(node);
  53.                     p.next = null; // help GC
  54.                     failed = false;
  55.                     return interrupted; 
  56.                 }
  57.                 if (shouldParkAfterFailedAcquire(p, node) &&
  58.                     parkAndCheckInterrupt())//线程进入等待状态
  59.                     interrupted = true;
  60.             }
  61.         } finally {
  62.             if (failed)
  63.                 cancelAcquire(node);
  64.         }
  65.     }
  67.      private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
  68.         int ws = pred.waitStatus;//前一个节点的状态
  69.         if (ws == Node.SIGNAL)
  70.             /*
  71.              * This node has already set status asking a release
  72.              * to signal it, so it can safely park.
  73.              */
  74.             return true;
  75.         if (ws > 0) {
  76.             /*
  77.              * Predecessor was cancelled. Skip over predecessors and
  78.              * indicate retry.
  79.              */
  80.             do {
  81.                 node.prev = pred = pred.prev;
  82.             } while (pred.waitStatus > 0);
  83.             pred.next = node;
  84.         } else {
  85.             /*
  86.              * waitStatus must be 0 or PROPAGATE.  Indicate that we
  87.              * need a signal, but don't park yet.  Caller will need to
  88.              * retry to make sure it cannot acquire before parking.
  89.              */
  90.             compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
  91.         }
  92.         return false;
  93.     }
  1.      public final boolean release(int arg) {
  2.             if (tryRelease(arg)) {
  3.                 Node h = head;
  4.                 if (h != null && h.waitStatus != 0)
  5.                     unparkSuccessor(h);
  6.                 return true;
  7.             }
  8.             return false;
  9.         }
  11.         private void unparkSuccessor(Node node) {
  12.         /*
  13.          * If status is negative (i.e., possibly needing signal) try
  14.          * to clear in anticipation of signalling.  It is OK if this
  15.          * fails or if status is changed by waiting thread.
  16.          */
  17.         int ws = node.waitStatus;
  18.         if (ws < 0)
  19.             compareAndSetWaitStatus(node, ws, 0);
  21.         /*
  22.          * Thread to unpark is held in successor, which is normally
  23.          * just the next node.  But if cancelled or apparently null,
  24.          * traverse backwards from tail to find the actual
  25.          * non-cancelled successor.
  26.          */
  27.         Node s = node.next;
  28.         if (s == null || s.waitStatus > 0) {
  29.             s = null;
  30.             for (Node t = tail; t != null && t != node; t = t.prev)
  31.                 if (t.waitStatus <= 0)
  32.                     s = t;
  33.         }
  34.         if (s != null)
  35.             LockSupport.unpark(s.thread);
  36.     }

具体流程图在Java并发编程的艺术P128页

共享式同步状态获取与释放

共享式同步状态获取与释放最主要的区别在于同一时刻能否有多个线程同时获取到同步状态。

  1.     //尝试获取同步状态,当返回值大于等于0时，表示可以获取同步状态
  2.      public final void acquireShared(int arg) {
  3.             if (tryAcquireShared(arg) < 0)
  4.                 doAcquireShared(arg);
  5.         }
  6.     //需要自己实现
  7.      protected int tryAcquireShared(int arg) {
  8.         throw new UnsupportedOperationException();
  9.     }
  11.     private void doAcquireShared(int arg) {
  12.         final Node node = addWaiter(Node.SHARED);//创建节点，并且添加到末尾，且返回节点
  13.         boolean failed = true;
  14.         try {
  15.             boolean interrupted = false;
  16.             for (;;) {
  17.                 final Node p = node.predecessor();//获取该节点前一个节点
  18.                 if (p == head) {//前一个节点为头节点
  19.                     int r = tryAcquireShared(arg);//获取同步状态
  20.                     if (r >= 0) {//获取成功
  21.                         setHeadAndPropagate(node, r);//修改头节点
  22.                         p.next = null; // help GC
  23.                         if (interrupted)
  24.                             selfInterrupt();
  25.                         failed = false;
  26.                         return;
  27.                     }
  28.                 }
  29.                 if (shouldParkAfterFailedAcquire(p, node) &&
  30.                     parkAndCheckInterrupt())
  31.                     interrupted = true;
  32.             }
  33.         } finally {
  34.             if (failed)
  35.                 cancelAcquire(node);
  36.         }
  37.     }

Condition的实现与分析

  1.     //初始节点
  2.     private transient Node firstWaiter;
  3.     //尾节点
  4.     private transient Node lastWaiter;
  6.     //将以当前线程构造节点，并将节点从尾部加入等待队列
  7.     //不需要CAS保证，因为调用await()方法的线程必定是获取了锁的线程
  8.     public final void await() throws InterruptedException {
  9.             if (Thread.interrupted())
  10.                 throw new InterruptedException();
  11.             Node node = addConditionWaiter();
  12.             int savedState = fullyRelease(node);
  13.             int interruptMode = 0;
  14.             while (!isOnSyncQueue(node)) {
  15.                 LockSupport.park(this);
  16.                 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
  17.                     break;
  18.             }
  19.             if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
  20.                 interruptMode = REINTERRUPT;
  21.             if (node.nextWaiter != null) // clean up if cancelled
  22.                 unlinkCancelledWaiters();
  23.             if (interruptMode != 0)
  24.                 reportInterruptAfterWait(interruptMode);
  25.     }

未完待续。。。

添加新批注
在作者公开此批注前，只有你和作者可见。
回复批注