JDK里的自旋锁

自旋锁是采用让当前线程不停地的在循环体内执行实现的，当循环的条件被其他线程改变时才能进入临界区。

JDK里面自旋锁的实现有 SynchronousQueue  和 LinkedTransferQueue。  本文只是自己对源码的简单理解。

先说公平锁，先等待的线程先获得数据。SynchronousQueue的内部类TransferQueue实现了公平锁。

某一时刻 线程A看到内存的情况如下：   链表，head 和 tail 分别指向链首和链尾，并且线程执行了ht = tail 。

*

* head ht=tail

* | |

* v v

* M - U - U - U

*

M 代表match ， 表示该节点已经匹配到了数据，  poll 等到了 offer ，或者 offer 等到了 poll 。

U 代表unmatch ， 表示该节点在等待填充数据或者数据等待poll 。

由于并发操作，经过其他线程的操作使得链表变成如下：（此操作是瞬间完成的）

* head ht tail

* | |  |

* v v v

* M - M1 - U - U - U1 - U2

*

有两个U节点(U1和U2)添加进来， 第二个节点U匹配到了数据，变成M1.

假设此时线程A想再往链表添加U节点：

那么需要进行那些操作？

1： 发现ht != tail , 于是需要继续循环

2：在 ht == tail 的情况下需要先 判断 tail.next 是否为null ， 如果不是则继续循环

3：新建一个node，将 tail.next 指向新node。

4：将U赋值给tail 。

* head tail

* | |

* v v

* U - U - U - U - Un

/*

 * head tail

 * | |

 * v v

 * U - U - U - U - Un

5：  自旋等待match。

6： 等待结束有两种可能结果： 6.1 超时取消    6.2 成功 

成功时： 此时自己的节点一定在第一位，因此只需将 head.next 指向node.next . 并且将 node.next 置空。

/*

 * head tail

 * | |

 * v v

 * M M M M M Mf - U - U - U - U - U


 */

等待超时：此时Ux按理也应该来到了链首，前面node的也都超时了，但万一其他的线程没有获得cpu呢，就会出现如下的状况，需要将U前面的几个node也顺便清理掉

/*

 * head tail

 * | |

 * v v

 * M M M M M U - U - Ux - U - U - U

 */

假设此时线程A是想将U变成M，这个逻辑很简单，按顺序找到一个U，尝试给U的item赋值。成功结束，不成功继续循环。

E transfer(E e, boolean timed, long nanos) {

 QNode s = null; // constructed/reused as needed

 boolean isData = (e != null);

 for (;;) {

 QNode t = tail;

 QNode h = head;

 if (t == null || h == null) // saw uninitialized value

 continue; // spin

 if (h == t || t.isData == isData) { // empty or same-mode

 QNode tn = t.next;

 if (t != tail) // inconsistent read

 continue;

 if (tn != null) { // lagging tail

 advanceTail(t, tn);

 continue;

 if (timed nanos = 0) // cant wait

 return null;

 if (s == null)

 s = new QNode(e, isData);

 if (!t.casNext(null, s)) // failed to link in

 continue;

 advanceTail(t, s); // swing tail and wait

 Object x = awaitFulfill(s, e, timed, nanos);

 if (x == s) { // wait was cancelled

 clean(t, s);

 return null;

 if (!s.isOffList()) { // not already unlinked

 advanceHead(t, s); // unlink if head

 if (x != null) // and forget fields

 s.item = s;

 s.waiter = null;

 return (x != null) ? (E)x : e;

 } else { // complementary-mode

 QNode m = h.next; // node to fulfill

 if (t != tail || m == null || h != head)

 continue; // inconsistent read

 Object x = m.item;

 if (isData == (x != null) || // m already fulfilled

 x == m || // m cancelled

 !m.casItem(x, e)) { // lost CAS

 advanceHead(h, m); // dequeue and retry

 continue;

 advanceHead(h, m); // successfully fulfilled

 LockSupport.unpark(m.waiter);

 return (x != null) ? (E)x : e;

}

自旋分析：

Object awaitFulfill(QNode s, E e, boolean timed, long nanos) {

 /* Same idea as TransferStack.awaitFulfill */

 final long deadline = timed ? System.nanoTime() + nanos : 0L;

 Thread w = Thread.currentThread();

 int spins = ((head.next == s) ?

 (timed ? maxTimedSpins : maxUntimedSpins) : 0);

 for (;;) {

 if (w.isInterrupted())

 s.tryCancel(e);

 Object x = s.item;

 if (x != e)

 return x;

 if (timed) {

 nanos = deadline - System.nanoTime();

 if (nanos = 0L) {

 s.tryCancel(e);

 continue;

 if (spins 0)

 --spins;

 else if (s.waiter == null)

 s.waiter = w;

 else if (!timed)

 LockSupport.park(this);

 else if (nanos spinForTimeoutThreshold)

 LockSupport.parkNanos(this, nanos);

}

循环判断 node.item 有没有变化，如果有变化则匹配成功，如果没有则继续循环， 循环一定的次数(spins)后还没有匹配成功则使用 LockSupport.park() 来阻塞线程。这个循环过程即可称为自旋。 如果等待时间为0，则spins为16，否则spins为16*32 .

以上可能多线程同时赋值的操作均采用了cas。

但是公平锁存在一个问题： 效率不够高。

/*
 *
 * head ht=tail
 * | |
 * v v
 * M - U - U - U
 * 
 * |
 * |
 * U变成M，但是此线程并没有获得cpu，没法立即执行，于是后面获得cpu的线程便有了意见，为啥不将数据给我，我能立即执行。
 */

要想解决此问题说来也十分简单，每个线程只需不停的将自己的node和head.next 进行交换即可优先获得task。 这个是在 TransferStack中实现的，说起来简单，但实现起来则困难得多。

原创文章，转载请注明： 转载自并发编程网 ifeve.com本文链接地址: JDK里的自旋锁

原创文章，作者：ItWorker，如若转载，请注明出处：https://blog.ytso.com/60297.html