AQS实现原理

AQS中维护了一个volatile int state(共享资源)和一个CLH队列。当state=1时代表当前对象锁已经被占用,其他线程来加锁时则会失败,失败的线程被放入一个FIFO的等待队列中,然后会被UNSAFE.park()操作挂起,等待已经获得锁的线程释放锁才能被唤醒。

我们拿具体场景来分析,假设同时有三个线程并发抢占锁,此时线程一抢占成功,线程二、三抢占失败,具体流程如下:

AQS实现原理

此时AQS内部数据结构为:

AQS实现原理

上图可以看到等待队列中的节点Node是一个双向链表,这里SIGNAL是Node中waitStatus属性。

以非公平锁看下具体实现:

java.util.concurrent.locks.ReentrantLock.NonfairSync:

static final class NonfairSync extends Sync { final void lock() { if (compareAndSetState(0, 1)) setExclusiveOwnerThread(Thread.currentThread()); else acquire(1); } protected final boolean tryAcquire(int acquires) { return nonfairTryAcquire(acquires); } }

线程进来直接利用CAS尝试抢占锁,如果抢占成功state值会被修改为1,且设置对象独占锁线程为当前线程。

线程抢占实现

线程二抢占失败,执行acquire(1)方法。

java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire()

public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }

tryAcquire是AbstractQueuedSynchronizer的方法,未提供对应实现,由子类实现:

java.util.concurrent.locks.ReentrantLock .nonfairTryAcquire()

final boolean nonfairTryAcquire(int acquires) { final Thread current = Thread.currentThread(); int c = getState(); if (c == 0) { if (compareAndSetState(0, acquires)) { setExclusiveOwnerThread(current); return true; } } else if (current == getExclusiveOwnerThread()) { int nextc = c + acquires; if (nextc < 0) throw new Error("Maximum lock count exceeded"); setState(nextc); return true; } return false; }

nonfairTryAcquire()方法中首先会获取state的值,如果不为0则说明当前对象的锁已经被其他线程占有,接着判断占有锁的线程是否为当前线程,如果是则累加state值,这里其实就是可重入锁的具体实现。如果state为0,则执行CAS操作,尝试更新state值为1,如果更新成功则代表当前线程加锁成功。

当前线程二执行tryAcquire()后返回false,接着执行addWaiter(Node.EXCLUSIVE)逻辑,将自己加入到一个FIFO等待队列中,代码实现如下:

java.util.concurrent.locks.AbstractQueuedSynchronizer.addWaiter()

private Node addWaiter(Node mode) { Node node = new Node(Thread.currentThread(), mode); Node pred = tail; if (pred != null) { node.prev = pred; if (compareAndSetTail(pred, node)) { pred.next = node; return node; } } enq(node); return node; }

此时队列中tail指针为空,直接调用enq(node)方法将当前线程加入等待队列尾部:

private Node enq(final Node node) { for (;;) { Node t = tail; if (t == null) { if (compareAndSetHead(new Node())) tail = head; } else { node.prev = t; if (compareAndSetTail(t, node)) { t.next = node; return t; } } } }

第一次循环时tail为空,创建一个哨兵节点,head指向这个哨兵节点;第二次循环,将线程二对应的node节点挂载到head节点后面并返回当前线程创建的节点信息。继续往后执行acquireQueued(addWaiter(Node.EXCLUSIVE), arg)逻辑,此时传入的参数为线程二对应的node节点信息。

java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireQueued()

final boolean acquireQueued(final Node node, int arg) { boolean failed = true; try { boolean interrupted = false; for (;;) { final Node p = node.predecessor(); if (p == head && tryAcquire(arg)) { setHead(node); p.next = null; // help GC failed = false; return interrupted; } if (shouldParkAfterFailedAcquire(p, node) && parkAndChecknIterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } } private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { int ws = pred.waitStatus; if (ws == Node.SIGNAL) return true; if (ws > 0) { do { node.prev = pred = pred.prev; } while (pred.waitStatus > 0); pred.next = node; } else { compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } return false; } private final boolean parkAndCheckInterrupt() { LockSupport.park(this); return Thread.interrupted(); }

内容版权声明:除非注明,否则皆为本站原创文章。

转载注明出处:https://www.heiqu.com/zgysgf.html