并发编程--AbstractQueuedSynchronizer介绍和原理分析

àì夳堔傛蜴生んèń 2022-06-18 12:08 162阅读 0赞

AbstractQueuedSynchronizer是并发编程包中最重要的类，是并发编程包的实现基层。简单来说，AbstractQueuedSynchronizer提供了一个基于FIFO的队列，用于存储处于阻塞状态的线程；提供了一个volatile修改的state变量，用于作为锁获取的标识位，对于FIFO队列和state的操作都是通过Unsafe这个类来实现CAS原子操作。

AQS的功能可以分为两类：独占功能和共享功能，它的所有子类中，要么实现了并使用了它独占锁功能的API，要么使用了共享锁的功能，而不会同时使用两套API，即便是它最有名的子类读写锁ReentrantReadWriteLock也是通过两个内部类：读锁和写锁，分别实现两套API来实现的。

### 一、FIFO队列： ###

//阻塞线程节点
        static final class Node {
    		
            static final Node SHARED = new Node();  //共享锁
    
            static final Node EXCLUSIVE = null;  //独占锁
    
            static final int CANCELLED =  1;  //用于表示当前线程被取消
    
            static final int SIGNAL    = -1;  //表示当前线程处于阻塞状态
    
            static final int CONDITION = -2;  //表示当前节点在等待Condition
     
            static final int PROPAGATE = -3;  //表示当前场景下后续的acquireShared能够得以执行
    
            volatile int waitStatus;
    
            volatile Node prev;
    
            volatile Node next;
    
           
            volatile Thread thread;
    
            Node nextWaiter;
    	}

当线程没有竞争到锁资源时就会被阻塞并放到FIFO队列中，当锁资源释放后再从FIFO由阻塞状态变为可执行状态来获取锁资源并执行。当线程获取锁资源时有公平锁和非公平锁可以参考 [并发编程--公平锁和非公平锁][--]；此外对于锁的分类还有独占锁和共享锁，简单来说独占锁state就是当前锁只能有一个线程获取，其他线程处于阻塞状态，共享锁是多个线程可以同时持有当前锁状态state，只要符合某种要求所有线程可以并发执行，不用进行阻塞等待。

AQS FIFO队列模型：

![20160413232805271][]

### 二、锁状态： ###

AbstractQueuedSynchronizer提供了一个变量state，用来作为当前锁是否可以获取的标识，对于所有线程来说，当前state的值对于他们来说都是可见的，每个线程持有的state值都是相同的。

private volatile int state;

### 三、获取锁 ###

#### 1、成功获取锁 ####

我们介绍一下一个线程是如何获取锁的，当一个独占锁被获取之后，其他线程是该如何进行操作的。我们用ReentrantLock的接口lock来看一下锁是如何获取的。

lock接口：

final void lock() {
    		//但获取锁时，会将state设置为1，如果是单个线程多次获取锁则state++
                acquire(1);
            }

（1）tryAcquire是尝试获取锁，如果获取锁则继续执行  
（2）如果tryAcquire返回false，则调用acquireQueued将当前线程阻塞并添加的FIFO队列中

//tryAcquire是尝试获取锁，如果获取锁则继续执行
    //如果tryAcquire返回false，则调用acquireQueued将当前线程阻塞并添加的FIFO队列中
    public final void acquire(int arg) {
            if (!tryAcquire(arg) &&
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
                selfInterrupt();
        }

我们看看tryAcquire尝试获取锁的机制，我们介绍一下公平锁的获取吧

（1）首先会获取lock中的state值，如果state等于0表示当前锁可获取，接下来hasQueuedPredecessors判断当前锁是否在FIFO队列中，当FIFO队列为空时，当前线程可以获取锁，不然根据公平规则，需要让FIFO中的线程优先获取锁，当可以获取锁是调用 compareAndSetState(0, acquires)将state值通过cas操作设置为acquires，当前线程获取了锁，state值为acquires值，其他线程来获取锁时state肯定就不为0了。

（2）当state值不为0时，首先要判断一下当前获取锁的线程和申请锁的线程是否是一个线程，如果时一个线程state++，当前线程获取了多次锁

（3）当以上条件都不满足时，表示线程无法获取锁，返回false。

protected final boolean tryAcquire(int acquires) {
    			//获取当前线程
                final Thread current = Thread.currentThread();
    			//获取锁标识state
                int c = getState();
    			//如果c等于0则表示当前线程可以获取锁
                if (c == 0) {
    				//hasQueuedPredecessors判断当前锁是否在FIFO队列中，当FIFO队列为空时，当前线程可以获取锁，不然根据公平规则，需要让FIFO中的线程优先获取锁
    				//当可以获取锁是调用 compareAndSetState(0, acquires)将state值通过cas操作设置为acquires
                    if (!hasQueuedPredecessors() &&
                        compareAndSetState(0, acquires)) {
    					//将独占线程设置为当前线程
                        setExclusiveOwnerThread(current);
                        return true;
                    }
                }
    			//即使当前获取不到锁，如果获取锁的线程是当前线程则state++，线程获取多个锁
                else if (current == getExclusiveOwnerThread()) {
                    int nextc = c + acquires;
                    if (nextc < 0)
                        throw new Error("Maximum lock count exceeded");
                    setState(nextc);
                    return true;
                }
    			//获取锁失败
                return false;
            }
        }

总结：锁的获取是根据锁状态值state和判断当前获取锁的线程是否是申请锁的线程来判断的

#### 2、进入FIFO线程队列 ####

当获取锁失败时返回false，则执行acquireQueued(addWaiter(Node.EXCLUSIVE), arg)操作。

首先调用addWaiter(Node.EXCLUSIVE)，将线程添加到FIFO队列中，并阻塞。

private Node addWaiter(Node mode) {
    		//创建当前线程节点Node
            Node node = new Node(Thread.currentThread(), mode);
    		
            Node pred = tail;
    		//当尾节点不为空时
            if (pred != null) {
                node.prev = pred;
    			//cas原子操作将当前线程添加到FIFO队列中
                if (compareAndSetTail(pred, node)) {
                    pred.next = node;
                    return node;
                }
            }
    		//当尾节点为空时，初始化head和tail节点
            enq(node);
    		//返回node
            return node;
        }

接下来调用acquireQueued将线程进行阻塞

final boolean acquireQueued(final Node node, int arg) {
            boolean failed = true;
            try {
                boolean interrupted = false;
                for (;;) {
    				//获取node节点的前一个节点
                    final Node p = node.predecessor();
    				//再尝试获取锁
                    if (p == head && tryAcquire(arg)) {
                        setHead(node);
                        p.next = null; // help GC
                        failed = false;
                        return interrupted;
                    }
    				//shouldParkAfterFailedAcquire 用来设置node的waitstatus状态为SIGNAL，值为-1，表示当前节点的后继节点包含的线程需要运行，也就是unpark；
    				//parkAndCheckInterrupt() 操作是将当前线程park，阻塞，线程阻塞在这个地方
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        interrupted = true;
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }

线程阻塞的操作是在parkAndCheckInterrupt中将当前线程阻塞。

private final boolean parkAndCheckInterrupt() {
    		//阻塞当前线程
            LockSupport.park(this);
            return Thread.interrupted();
        }

总结：以上操作完成了线程的进FIFO队列及当前线程的阻塞。

### 3、释放锁资源 ###

锁的释放还是比较简单的，简单的来说就是将锁标识位state进行减一操作，然后将阻塞线程唤起让他们重新竞争

public void unlock() {
            sync.release(1);
        }

release中进行state恢复值并唤起阻塞线程。

public final boolean release(int arg) {
    		//tryRelease释放锁，将state进行减值，并设置当前可执行线程为null
            if (tryRelease(arg)) {
                Node h = head;
                if (h != null && h.waitStatus != 0)
    				//唤起阻塞线程
                    unparkSuccessor(h);
                return true;
            }
            return false;
        }

tryRelease中对state进行修改，如果state的值为0的话将当前执行线程设置为null。

protected final boolean tryRelease(int releases) {
    			//修改state值
                int c = getState() - releases;
                if (Thread.currentThread() != getExclusiveOwnerThread())
                    throw new IllegalMonitorStateException();
                boolean free = false;
                if (c == 0) {
                    free = true;
    				//将当前执行线程设置为空
                    setExclusiveOwnerThread(null);
                }
    			//修改state值
                setState(c);
                return free;
            }

接下来的操作是调用unparkSuccessor将阻塞线程唤起

private void unparkSuccessor(Node node) {
            
            int ws = node.waitStatus;
            if (ws < 0)
                compareAndSetWaitStatus(node, ws, 0);
    		
            Node s = node.next;
            if (s == null || s.waitStatus > 0) {
                s = null;
                for (Node t = tail; t != null && t != node; t = t.prev)
                    if (t.waitStatus <= 0)
                        s = t;
            }
    		//唤起当前线程
            if (s != null)
                LockSupport.unpark(s.thread);
        }

总结：AbstractQueuedSynchronizer 简单来说设置了一个锁标识state，通过判断state的值来确定是否线程可以获取锁，如果无法获取锁则将当前线程添加的FIFO队列中并进行阻塞，当线程使用完锁之后就会释放锁资源（修改state值），然后唤起FIFO队列中的阻塞线程来竞争获取锁。

[--]: http://blog.csdn.net/qq924862077/article/details/70161631
[20160413232805271]: /images/20220618/0157346302cc4158aba53f33f973e52a.png