java多线程 - AbstractQueuedSynchronizer的理解

红太狼 2024-04-17 06:30 71阅读 0赞

AbstractQueuedSynchronizer在JDK1.8的实现类如下：

尝试着从ReentrantLock的功能出来，来理解AQS的功能点。

### AQS ###

站在使用者的角度来看AQS，AQS可以分为两种模式：  
(1) exclusive mode 独占功能  
(2) share mode 共享功能

ReentrantLock使用了AQS的独占功能

reentrantLock.lock();
       //do something
    reentrantLock.release();
    复制代码

ReentrantLock会保证dosomething的线程在同一时刻只有一个，其他线程会被挂起，直到获取锁。从这里可以看出，其实ReentrantLock实现的就是一个独占锁的功能：有且只有一个线程获取到锁，其余线程全部挂起，直到该拥有锁的线程释放锁，被挂起的线程被唤醒重新开始竞争锁。没错，ReentrantLock 使用的就是 AQS 的独占 API 实现的

重点看一下ReentrantLock是如何获取锁的：

### 加锁 ###

/**
         * Acquires the lock.
         *
         * <p>Acquires the lock if it is not held by another thread and returns
         * immediately, setting the lock hold count to one.
         *
         * <p>If the current thread already holds the lock then the hold
         * count is incremented by one and the method returns immediately.
         *
         * <p>If the lock is held by another thread then the
         * current thread becomes disabled for thread scheduling
         * purposes and lies dormant until the lock has been acquired,
         * at which time the lock hold count is set to one.
         */
        public void lock() {
            sync.lock();
        }
    复制代码

可以简单明了的看出来，reentrantlock在加锁的时候，仅仅是调用了sync的的lock方法，这个内部类维护在reentrantlock，分为公平与非公平。

公平锁：每个线程抢占锁的顺序为先后调用lock方法的顺序依次获取锁，类似于排队吃饭

非公平锁：每个线程抢占锁的顺序不定，谁运气好，谁就获取到锁，和调用 lock 方法的先后顺序无关

先看一下ReentrantLock的公平锁的实现方式：

final void lock() {
         acquire(1);
     }
    复制代码

调用了AQS的acquire方法,从代码的语义上可以大概能看出来，尝试获取锁，如果获取不到便创建一个waiter（当前线程）放入到队列中

回形针1

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

AQS提供了一个空壳子，供子类实现：
    
        protected boolean tryAcquire(int arg) {
           throw new UnsupportedOperationException();
        }
    复制代码

公平锁的实现方式：

protected final boolean tryAcquire(int acquires) {
                final Thread current = Thread.currentThread();
                int c = getState();
                if (c == 0) {
                    if (!hasQueuedPredecessors() &&
                        // 查看队列中是否有比当前线程排在前面的线程
                        compareAndSetState(0, acquires)) {
                        // 使用CAS修改状态    
                        setExclusiveOwnerThread(current);
                        return true;
                    }
                }
                else if (current == getExclusiveOwnerThread()) {
                    // 如果状态不等于0，便代表锁已经被获取，由于reentantLock是可重入锁，因此如果当前锁的线程与现在相同，将计数器+1
                    int nextc = c + acquires;
                    if (nextc < 0)
                        throw new Error("Maximum lock count exceeded");
                    setState(nextc);
                    return true;
                }
                return false;
            }
    复制代码

上步有个获取状态的方法。这个状态维护在AQS中，以来表述当前锁的状态

思路继续回到 -> 回形针1，如果线程获取锁失败，便会将这个线程进行一些处理之后，放入队列中

private Node addWaiter(Node mode) {
            Node node = new Node(Thread.currentThread(), mode);
            // Try the fast path of enq; backup to full enq on failure
            Node pred = tail;
            if (pred != null) {
                node.prev = pred;
                if (compareAndSetTail(pred, node)) {
                    pred.next = node;
                    return node;
                }
            }
            enq(node);
            return node;
        }
    复制代码

用当前线程去构建一个Node对象。可以看出，在AQS的这个队列中中可以明确区分哪些是独占模式，哪些是共享模式。

创建好节点之后，将节点加入到队列的尾部。此处，在队列不为空的时候，先尝试使用cas方式修改尾部节点为最新的节点，如果修改失败，意味着有并发，这时候才会进入到enq中死循环，自旋的方式修改。

将线程放入到链表中，还有一步比较重要的是：将线程挂起 （该步骤具体由 \#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) &&
                        parkAndCheckInterrupt())
                        interrupted = true;
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }
    复制代码

对线程的挂起及唤醒操作是通过使用 UNSAFE 类调用 JNI 方法实现的

### 释放锁 ###

释放锁，成功后，找到 AQS 的头节点，并唤醒它即可

protected final boolean tryRelease(int releases) {
                int c = getState() - releases;
                if (Thread.currentThread() !=getExclusiveOwnerThread())
                    throw new IllegalMonitorStateException();
                boolean free = false;
                // 因为reentrantLock是重入的，所以不是每次释放锁都是等于0的
                if (c == 0) {
                    free = true;
                    setExclusiveOwnerThread(null);
                }
                setState(c);
                return free;
            }
    复制代码

ReentrantLock的lock和unlock方法已经基本解析完毕了，唯独还剩下一个非公平锁 NonfairSync没说。其实，它和公平锁的唯一区别就是获取锁的方式不同，一个是按前后顺序一次获取锁，一个是抢占式的获取锁

非公平锁的 lock 方法的处理方式是: 在 lock 的时候先直接 cas 修改一次 state 变量（尝试获取锁），成功就返回，不成功再排队，从而达到不排队直接抢占的目的。

而对于公平锁：则是老老实实的开始就走 AQS 的流程排队获取锁。如果前面有人调用过其 lock 方法，则排在队列中前面，也就更有机会更早的获取锁，从而达到“公平”的目的。

### 总结 ###

总的来说，思路其实并不复杂，还是使用的标志位+队列的方式，记录获取锁、竞争锁、释放锁等一系列锁的状态。从AQS的层面state可以表示锁，也可以表示其他状态，它并不关心它的子类把它变成一个什么工具类，而只是提供了一套维护一个独占状态。AQS只是维护一个状态，一个控制各个线程何时可以访问的状态，它只对状态负责，而这个状态表示什么含义，由子类自己去定义。

转载于:https://juejin.im/post/5d352ac9e51d454f73356e0f