Java并发：AbstractQueuedSynchronizer

淩亂°似流年 2022-12-01 04:00 138阅读 0赞

AbstractQueuedSynchronizer：抽象同步队列，简称AQS，它是实现同步器的基础组件，并发包中的锁的底层就是使用AQS实现的。

**开始正题：**  
先看下AQS中存在的成员变量

**成员变量：**

//队列头结点
        private transient volatile Node head;
        //队列尾节点
        private transient volatile Node tail;
        // 同步器的状态，例如ReentrantLock用作锁是否被占用，锁重入的次数
        // CountDownLatch用于记录次数 
        private volatile int state;

**Node节点**

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;
            static final int PROPAGATE = -3;
           /** * 表示节点的状态。其中包含的状态有： * CANCELLED，值为1，表示当前的线程被取消； * SIGNAL，值为-1，表示当前节点的后继节点包含的线程需要运行，也就是unpark； * CONDITION，值为-2，表示当前节点在等待condition，也就是在condition队列中； * PROPAGATE，值为-3，表示当前场景下后续的acquireShared能够得以执行； * 值为0，表示当前节点在sync队列中，等待着获取锁。 */
            volatile int waitStatus;
            // 前驱节点
            volatile Node prev;
            // 后继节点
            volatile Node next;
            // 入队的线程
            volatile Thread thread;
            // Condition中使用的
            Node nextWaiter;
    		...一众修改的方法

现在为止，我们可以获得的信息，AQS维护着一个双向队列，队列的节点为Node，Node保存的信息有prev，next，waitStatus，Thread，节点模式。

下面我们就用ReentrantLock的不公平模式来对AQS进行剖析：  
**成员变量：**

// Sync是ReentrantLock的静态内部类，继承AbstractQueuedSynchronizer。
    private final Sync sync;

**构造方法：**

public ReentrantLock() { 
        	//NonfairSync是不公平模式的实现
            sync = new NonfairSync();
        }

**Lock：**

//获取锁方法
        final void lock() { 
             // ReentrantLock将AQS的变量state用来表示锁的状态
             // 0表示锁的空闲，1表示锁已经被占有，2~n表示锁的重入次数
             // 使用Unsafe来设置state，设置0到1成功则表示获取锁
             // 尝试获取锁
             if (compareAndSetState(0, 1))
                 // 设置获取锁的线程，AbstractOwnableSynchronizer的方法，用来复制AQS
                 // 便于记录获取锁的线程
                 setExclusiveOwnerThread(Thread.currentThread());
              else
                 // 尝试获取
                 acquire(1);
        }
        //AbstractQueuedSynchronizer类
        public final void acquire(int arg) { 
            // 尝试获取锁，考虑锁空闲或者锁重入情况
            // 获取锁失败后运行下个方法
            if (!tryAcquire(arg) &&
                // addWaiter(Node.EXCLUSIVE), arg)：将当前线程加入sync队列中
                // addWaiter返回new Node(Thread.currentThread(), Node.EXCLUSIVE)节点
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))         
                // 中断线程
                selfInterrupt();
        }
        // 尝试获取锁
        protected final boolean tryAcquire(int acquires) { 
            return nonfairTryAcquire(acquires);
        }
        // 不公平模式下的锁尝试获取
        final boolean nonfairTryAcquire(int acquires) { 
        		// 获取当前线程
                final Thread current = Thread.currentThread();
                // 获取锁的状态
                int c = getState();
                // state==0，锁空闲
                if (c == 0) { 
                	// 尝试cas修改state的值
                    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;
          }
          // 队列中添加节点
          private Node addWaiter(Node mode) { 
            // 构建独占模式Node.EXCLUSIVE的node节点，节点会方法结束后返回
            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;
        }
        // 入队
        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;
                    }
                }
            }
        } 
        //参数：(new Node(Thread.currentThread(), Node.EXCLUSIVE)，1)
        // 这是一个死循环方法
        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)) { 
                    	// 成功获取锁之后的操作
                        // 也就是把node节点踢出队列并设置为头结点
                        setHead(node);
                        p.next = null; 
                        failed = false;
                        return interrupted;
                    }
                    //靠前继节点判断当前线程是否应该被阻塞和挂起线程，检查是否中断
                    if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
                        interrupted = true;
                }
            } finally { 
                if (failed)
                    cancelAcquire(node);
            }
        }
        // 根据waitStatus判断是否挂起
        private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { 
        	// 获取前驱节点的waitStatus，初始值的0
            int ws = pred.waitStatus;
            // 设置前驱节点的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() { 
        	// 这里是线程挂起，
        	// 当线程又被唤醒后，判断是否中断，没有中断则又会回到acquireQueued方法中
        	// 
            LockSupport.park(this);
            return Thread.interrupted();
        }

**流程图：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMDA4NTM1_size_16_color_FFFFFF_t_70_pic_center]  
**unLock：**

public void unlock() { 
            sync.release(1);
        }
        //AbstractQueuedSynchronizer
        public final boolean release(int arg) { 
        	// 尝试释放锁
            if (tryRelease(arg)) { 
           		// 指向头结点
                Node h = head;
                // 队列不为空 并且 waitStatus != 0
                if (h != null && h.waitStatus != 0)
                	
                    unparkSuccessor(h);
                return true;
            }
            return false;
        }
    	//尝试释放锁
    	protected final boolean tryRelease(int releases) { 
            //获取释放锁后的状态
            int c = getState() - releases;
            //判断线程是否为当前线程
            if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
            boolean free = false;
            // 锁的重入次数释放完了
            if (c == 0) { 
                free = true;
                setExclusiveOwnerThread(null);
            }
            setState(c);
            return free;
        }
    	// 释放队列第一个线程
    	// 传入队列头结点
    	private void unparkSuccessor(Node node) { 
    		//获取状态
            int ws = node.waitStatus;
            if (ws < 0)
            	// 修改状态
                compareAndSetWaitStatus(node, ws, 0);
    		//获取节点的后继节点
            Node s = node.next;
            // 头结点的后继节点为null 或者 waitStatus = CANCELLED
            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;
            }
            //唤醒该线程
            // 唤醒线程之后又回到了acquireQueued方法的的parkAndCheckInterrupt()方法里面
            if (s != null)
            	// 线程唤醒之后，会再尝试获取锁一次，没有获取到锁则再次挂起
                LockSupport.unpark(s.thread);
        }

**流程图：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMDA4NTM1_size_16_color_FFFFFF_t_70_pic_center 1]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMDA4NTM1_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/dbc76aba18514fc29ee82322ea5d6d67.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMDA4NTM1_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/beeffe98c2c7435e917207a548007be0.png