AbstractQueuedSynchronizer源码分析

简介

AbstractQueuedSynchronizer(下文均以AQS代替)可以用于构建锁或者其他相关同步装置的基础框架.JDK1.5之前一般使用synchronized关键字来实现线程对共享变量的互斥访问,而JDK1.5之后开发了AQS组件,使用原生java代码实现了synchronized语义.
AQS实现了两种模式：公平模式和非公平模式,在我们实现自定义锁的时候(后续会自己实现锁)只需按照要求即可实现对应模式或两种模式都实现,ReentrantLock就是实现了上述两种模式的锁
在使用公平模式时,AQS会生成一个FIFO队列,严格按照线程挂起顺序获得锁使用权.或者说AQS提供了FIFO队列
AQS内部包含了两个内部类,可以先从这个简单的内部类开始解析

Node类解析

Node是用来构成FIFO队列的基本单元,当前线程获取同步状态失败后,AQS会将当前线程以及等待状态等信息构造成节点并加入到同步队列中,同时会阻塞当前线程,当同步状态释放后,会把首节点的线程唤醒,使其再次尝试获取同步状态
它是用来保存获取同步状态失败的线程引用和线程状态、等待状态以及前驱和后继节点的容器,用来实现线程阻塞.
具体代码解析如下:

static final class Node {
	/** static类的属性 */
	/** 指示节点在共享模式下等待的标记 */
	static final Node SHARED = new Node();
	/** 指示节点在独占模式下等待的标记 */
	static final Node EXCLUSIVE = null;

	/** waitStatus值以指示线程已取消 */
	static final int CANCELLED =  1;
	/** waitStatus 值以指示后续线程需要 unparking */
	static final int SIGNAL    = -1;
	/** waitStatus 值以指示线程正在等待条件 */
	static final int CONDITION = -2;
	/**
	 * waitStatus 值以指示下一个 acquireShared 应无条件传播
	 */
	static final int PROPAGATE = -3;

	/**
	 * 表示节点的状态字段.仅接受以下值:
	 *   SIGNAL: 值为-1,线程的后继线程正/已被阻塞处于等待状态,
	 * 			当该线程释放了同步状态或取消时,将会通知后继节点,
	 *			使后继节点的线程得以运行(unpark)
	 *   CANCELLED: 值为1,由于在同步队列中等待的线程等待超时或者被中断
	 *			需要从同步队列中取消等待,节点进入该状态不会变化
	 *   CONDITION: 值为-2,节点在等待队列中,节点线程等待在Condition上,
	 * 			当其他线程对Condition调用了singal方法后,该节点将会从等待队列中
	 * 			转移到同步队列中,加入到同步状态的获取中
	 *   PROPAGATE: 值为-3,表示下一次共享模式同步状态获取将会无条件地传播下去
	 *   INITIAL: 初始状态值为0
	 */
	volatile int waitStatus;

	/**
	 * 链接到前驱节点,当前节点/线程依赖它来检查waitStatus
	 * 在入同步队列时被设置,并且仅在移除同步队列时才归零(便于GC)
	 */
	volatile Node prev;

	/**
	 * 链接到后继节点,当前节点/线程释放时释放.在入同步队列期间分配
	 * 在绕过取消的前驱节点时调整,并在出同步队列时取消
	 */
	volatile Node next;

	/**
	 * 将节点入列的线程,构造时初始化,使用后清零
	 */
	volatile Thread thread;

	/**
	 * 链接到下一个节点等待条件,或者特殊值SHARED
	 * 因为条件队列只有在保持在独占模式时才会被访问
	 * 所以只需要一个简单的链接队列来保持节点,同时等待条件
	 * 然后将它们转移到队列中以重新获取.并且因为条件只能是排它的
	 * 我们通过使用特殊的值来指示共享模式来保持一个字段
	 */
	Node nextWaiter;

	/**
	 * 如果节点在共享模式下等待,则返回true
	 */
	final boolean isShared() {
		return nextWaiter == SHARED;
	}

	/**
	 * 返回上一个节点,为null则抛出NullPointerException
	 * 当前驱节点不为null时使用
	 */
	final Node predecessor() throws NullPointerException {
		Node p = prev;
		if (p == null)
			throw new NullPointerException();
		else
			return p;
	}

	// 用于建立初始头或共享标记
	Node() {    
	}

	// 由addWaiter使用
	Node(Thread thread, Node mode) {     
		this.nextWaiter = mode;
		this.thread = thread;
	}

	// 由Condition使用
	Node(Thread thread, int waitStatus) { 
		this.waitStatus = waitStatus;
		this.thread = thread;
	}
}

ConditionObject类解析

已在Condiiton源码解析中阐述,如有需要请看ConditionObject类解析连接
AQS的内部类Node和ConditionObject已经解析完毕,接下来解析AQS的成员变量,基础部分就解析完毕,就可以各种AQS相关的操作流程

AQS成员变量

共享资源与同步队列

/**
 * 同步队列的head,懒加载  
 * 除了初始化之外只能通过setHead来设置
 * 如果head存在,则它的waitStatus保证不是CANCELLED
 */
private transient volatile Node head;

/**
 * 同步队列的tail,懒加载.
 * 只能通过enq()方法或新增新节点的方式改变tail
 */
private transient volatile Node tail;

/**
 * 同步状态
 */
private volatile int state;

由上概述的成员变量可知:AQS维护了volatile int state(代表共享资源,同步器的状态值,如:1-已占有,0-未占有,>1-表示重入的数量)和一个FIFO线程等待队列;AQS使用了state状态位+FIFO排列方式记录了锁的获取和释放

state的访问方式主要是以下三种:

/**
 * 返回同步状态的当前值
 */
protected final int getState() {
	return state;
}

/**
 * 设置同步状态的值
 */
protected final void setState(int newState) {
	state = newState;
}

/**
 * 如果当前值=期望值,则原子性的将同步状态设置为给定更新值
 * 成功则返回true,失败则表示当前实际值不等于期望值
 */
protected final boolean compareAndSetState(int expect, int update) {
	return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

CAS支持变量

/**
 * 为了支持compareAndSet,我们需要在AQS中实现这一点:
 * 但是为了后期的增强,我们不用显示子类AtomicInteger,虽然它是有效和有用的
 * 作为迷你实现,我们原生的实现CAS的常用API,
 * 我们在处理的同时,对CASable字段执行相同操作
 */
private static final Unsafe unsafe = Unsafe.getUnsafe();
private static final long stateOffset;
private static final long headOffset;
private static final long tailOffset;
private static final long waitStatusOffset;
private static final long nextOffset;

内部类和成员变量等基础解析完毕,接下来就是AQS操作:

AQS操作解析

由上图可知:AQS实现了两种资源共享方式:独占模式+共享模式,独占对外提供的方法是acquire-release;共享对外提供方法是acquireShared-releaseShared;接下来分别就这两种方式进行阐述

AQS独占方式解析

acquire获取资源

/**
 * 以独占模式获取共享资源,忽略中断
 * 通过至少调用一次tryAcquire来实现,若成功则返回;
 * 否则线程将进入队列排队,可能会反复阻塞和解除阻塞,直到tryAcquire直到成功
 * 
 * 这个方法可用于实现Lock锁的lock方法
 */
public final void acquire(int arg) {
	if (!tryAcquire(arg) &&
		acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
		selfInterrupt();
}

tryAcquire方法

此方法尝试获取资源,AQS中代码如下:

1
2
3

protected boolean tryAcquire(int arg) {
	throw new UnsupportedOperationException();
}

AQS中的tryAcquire方法直接抛出异常,并没有实现具体的功能,具体的功能交给具体自定义同步器去实现了.
注意:此处并没有定义成abstract,因为AQS内部实现了独占模式和共享模式,不定义成abstract,自定义同步器可根据自己需求单独实现其中一种或两种都实现
如JDK内部的ReentrantLock已经使用FairSync和NonFairSync这两种方式帮我们实现了tryAcquire
FairSync的tryAcquire:

protected final boolean tryAcquire(int acquires) {
	// 获取当前线程
    final Thread current = Thread.currentThread();
    // 获取独占锁的状态
    int c = getState();
    // 为0意味着锁没有被任何线程所拥有
    if (c == 0) {
    	// 若锁没有被任何线程所拥有,则判断当前线程是否是CLH队列中的第一个线程,
    	// 若是的话,则获取该锁,设置锁的状态,并设置锁的拥有者为当前线程
        if (!hasQueuedPredecessors() &&
            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;
}

NonFairSync的tryAcquire:

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) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

addWaiter()

当tryAcquire尝试去获取锁的时候成功则返回true,失败则返回false;此时!tryAcquire就是true就会往下执行acquireQueued(addWaiter())
addWaiter(Node.EXCLUSIVE)的作用:创建”当前线程”的Node节点,且Node中记录”当前线程”对应的锁是”独占锁”类型,并且该节点添加到CLH队列的末尾

private Node addWaiter(Node mode) {  
    // 新建一个Node节点,节点对应的线程是"当前线程","当前线程"的锁的模型是mode
    Node node = new Node(Thread.currentThread(), mode);  
    Node pred = tail;  
    // 若CLH队列不为空,则将"当前线程"添加到CLH队列末尾  
    if (pred != null) {  
        node.prev = pred;  
        if (compareAndSetTail(pred, node)) {  
            pred.next = node;  
            return node;  
        }  
    }  
    // 若CLH队列为空,则调用enq()新建CLH队列,然后再将”当前线程”添加到CLH队列中。  
    enq(node);  
    return node;  
}

compareAndSetTail也属于CAS函数,也是通过”本地方法”实现的compareAndSetTail(expect, update)会以原子的方式进行操作,它的作用是判断CLH队列的队尾是不是为expect,是的话就将队尾设为update

1
2
3

private final boolean compareAndSetTail(Node expect, Node update) {  
    return unsafe.compareAndSwapObject(this, tailOffset, expect, update);  
}

enq()的作用很简单,如果CLH队列为空,则新建一个CLH表头;然后将node添加到CLH末尾,否则直接将node添加到CLH末尾

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

acquireQueued()

我们已经将当期线程添加到CLH队列中了.而acquireQueued的作用就是逐步的去执行CLH队列的线程,如果当前线程获取到了锁则返回,否则当前线程休眠直到唤醒并重新获取锁才返回;获取过程中head、tail不断移动将已经完成的队列节点清除

final boolean acquireQueued(final Node node, int arg) {  
    boolean failed = true;  // 标志是否成功拿到资源
    try {  
        // interrupted表示在CLH队列的调度中,"当前线程"在休眠时,有没有被中断过
        boolean interrupted = false;  
		// 自旋操作
        for (;;) {  
            // 获取上一个节点.node是"当前线程"对应的节点,这里就意味着”获取上一个等待锁的线程”
            final Node p = node.predecessor();  
			// 如果前驱节点是head,该节点则是阻塞的第一个节点,可以尝试去获取资源(前驱唤醒或interrupt)
            if (p == head && tryAcquire(arg)) {  
				// 获取到资源后将head指向该节点;即head指向节点是获取资源的节点或null
                setHead(node);
				// p的prev和next都被设置为null,方便GC将head之前的节点,即拿完资源节点全部出队  
                p.next = null; 
                failed = false;  
                return interrupted;  
            }  
			// 如果可以休息了,就进入waiting状态,直到被unpark()
            if (shouldParkAfterFailedAcquire(p, node) &&  
                parkAndCheckInterrupt())  
				// waiting状态被interrupt的,在标记中断为true
                interrupted = true;  
        }  
    } finally {  
        if (failed)  
            cancelAcquire(node);  
    }  
}

1.shouldParkAfterFailedAcquire
此方法检测和更新获取失败节点的状态,避免存在节点是取消或被中断了,确保当前节点是真的可以park了

/**
 * 检查和更新获取资源失败的节点状态
 * 如果线程阻塞则返回true 
 */
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
	int ws = pred.waitStatus;
   /**
	* 前驱节点已经设置了通知信号状态,Node节点可以休息
	*/
	if (ws == Node.SIGNAL)
		return true;
   /**
	* 前驱节点被取消了,跳过被取消节点,让上上节点与当前节点连接
	* 一直循环往前找,直到有waitStatus状态正常等待,就把当前节点与之连接
	*/	
	if (ws > 0) {
		do {
			node.prev = pred = pred.prev;
		} while (pred.waitStatus > 0);
		pred.next = node;
	} else {
		/*
		 * 当waitStatus状态是0或PROPAGATE,将前驱的状态设置为SIGNAL信号
		 * 告诉它拿完号通知下自己,也可能会失败
		 */
		compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
	}
	return false;
}

2.parkAndCheckInterrupt()
真正让线程进入waiting状态,进行park的操作;并进行检查中断的操作

private final boolean parkAndCheckInterrupt() {
	// 调用park()使线程进入waiting状态
	LockSupport.park(this);
	// 如果被唤醒查看是不是被中断了
	return Thread.interrupted();
}

parkAndCheckInterrupt()的作用是阻塞当前线程,并且返回”线程被唤醒之后”的中断状态.它会先通过LockSupport.park()阻塞”当前线程”,然后通过Thread.interrupted()返回线程的中断状态

线程被阻塞之后如何唤醒:
第1种情况：unpark()唤醒.”前继节点对应的线程”使用完锁之后,通过unpark()方式唤醒当前线程
第2种情况：中断唤醒.其它线程通过interrupt()中断当前线程

补充: LockSupport()中的park(),unpark()的作用和 Object中的wait(),notify()作用类似,是阻塞/唤醒;它们的用法不同,park(),unpark()是轻量级的,而wait(),notify()是必须先通过Synchronized获取同步锁
3.acquireQueue的流程图

acquire流程总结

1.调用tryAcquire()尝试去获取资源,成功则返回
2.失败则addWaiter将线程对应节点放到队列尾部,并标记为独占模式,
3.acquireQueued是不是第一个在等待的,是就再尝试获取,不行就去休息
4.等待过程如果被中断,它不响应,等待获取资源即当前线程在活跃后再去处理

release释放资源

独占模式下资源的释放,也是Lock锁的unlock的具体实现

/**
 * 独占模式下进行资源释放  
 * true则释放一个或多个线程来实现
 */
public final boolean release(int arg) {
	if (tryRelease(arg)) {
		Node h = head;
		if (h != null && h.waitStatus != 0)
			unparkSuccessor(h);
		return true;
	}
	return false;
}

1.tryRelease尝试释放

1
2
3

protected boolean tryRelease(int arg) {
	throw new UnsupportedOperationException();
}

同样此处也是直接抛出异常的,具体代码由实现类去实现,可参考ReentrantLock的Sync,已经重写了tryRelease,具体代码如下:

protected final boolean tryRelease(int releases) {
	// 状态值-1
    int c = getState() - releases;
	// 当前线程与锁持有的线程不一致则抛出异常
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
	// c==0则所有的锁都unlock,将free置为true,独占锁置为null
    if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}

2.unparkSuccessor

若释放锁成功后,遍历节点将需要取消的取消,然后再次尾部开始查找最前边的未取消的节点并进行唤醒,尝试争夺资源

private void unparkSuccessor(Node node) {
    int ws = node.waitStatus;
    //如果waitStatus < 0 则将当前节点清零
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    //若后续节点为空或已被cancel，则从尾部开始找到队列中最前边的waitStatus<=0，即未被cancel的节点
    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);// 唤醒
}

用unpark()唤醒队列最前边的未放弃的节点,就会去执行acquire()去尝试获取资源,它已经队列最前边的未放弃节点了,即它已经在占用节点后面的第一个等待节点了,即使这一次获取失败也会前驱节点设置为SIGNAL,然后下次就会获取成功

AQS共享方式解析

acquireShared(int)

共享模式下获取资源的方法,

/**
 * 共享模式下获取资源,不响应中断  
 * 期间至少要调用一次tryAcquireShared,成功则返回,否则线程被阻塞
 * 可能反复阻塞和解阻塞直到成功  
 */
public final void acquireShared(int arg) {
	if (tryAcquireShared(arg) < 0)
		doAcquireShared(arg);
}

1.tryAcquireShared

/**
 * 共享模式下尝试获取资源
 * 这方法前提得查询对象是否允许在共享模式中获取它
 * 成功则返回,失败则对线程节点进行排队,直到收到SIGNAL信号
 */
protected int tryAcquireShared(int arg) {
	throw new UnsupportedOperationException();
}

又看到直接抛出异常的方法,大家应该知道此方法肯定是由实现类去具体实现了,这里后续每一个实现类都会进行源码分析,咱后续再讨论;
该方法已经明确了返回值:

失败则返回负值
在共享模式下获取成功,但是没有后续资源可以获取则返回0
在共享模式下获取成功,并且后续获取资源也成功则返回正值,这种情况下,后续等待线程必须检查可用性

2.doAcquireShared

/**
 * 在共享不中断模式下获取资源
 */
private void doAcquireShared(int arg) {
	// 加入队列尾部
	final Node node = addWaiter(Node.SHARED);
	// 是否成功标志
	boolean failed = true;
	try {
		// 等待过程是否被中断标志
		boolean interrupted = false;
		for (;;) {
			// 获取到前驱节点
			final Node p = node.predecessor();
			// 前驱节点是head,则当前节点可以通过head唤醒,从而去获取资源
			if (p == head) {
				// 尝试获取资源
				int r = tryAcquireShared(arg);
				// 获取成功
				if (r >= 0) {
					// 将head指向自己,并进行传播(还有资源可供后面线程唤醒使用)
					setHeadAndPropagate(node, r);
					p.next = null; // help GC
					// 此处处理中断
					if (interrupted)
						selfInterrupt();
					failed = false;
					return;
				}
			}
			// 判断状态,寻找可以park点进行waiting
			if (shouldParkAfterFailedAcquire(p, node) &&
				parkAndCheckInterrupt())
				interrupted = true;
		}
	} finally {
		if (failed)
			cancelAcquire(node);
	}
}

与acquireQueued()唯一区别是中断处理操作selfIntertupt()放到doAcquireShared()里面,而前面的独占模式acquireQueued则是放在外面最后处理;即acquireQueued是先拿到资源后判断中断,而共享模式是在拿资源的过程中可以中断以便以后线程可以获取资源,关键就在于拿资源过程中是否可以被抢占
与独占模式相比,共享模式下必须是head节点后面的第一个等待节点去获取,共享资源有剩余的情况下就会去唤醒其他线程节点;而独占模式下,同一时刻只有一个线程在执行

/**
 * 设置队列的head节点,并在共享模式下检查后继节点是否在等待
 */
private void setHeadAndPropagate(Node node, int propagate) {
	Node h = head; // 记录之前的head节点
	setHead(node); // 设置头节点
	/*
	 * 如果还有剩余量或h为null或h被取消/被中断则唤醒下一个线程
	 */
	if (propagate > 0 || h == null || h.waitStatus < 0 ||
		(h = head) == null || h.waitStatus < 0) {
		Node s = node.next;
		if (s == null || s.isShared())
			doReleaseShared();
	}
}

releaseShared

/**
 * 在共享模式下释放资源  
 * 通过解除阻塞一个或多个线程来实现
 */
public final boolean releaseShared(int arg) {
    // 尝试释放资源
	if (tryReleaseShared(arg)) {
		// 唤醒后继节点
		doReleaseShared();
		return true;
	}
	return false;
}

即先释放掉资源后唤醒后继,与独占模式下release一致,但是独占模式下tryRelease在完全释放掉资源后才会去唤醒其他节点,主要是独占下可重入的问题,而共享模式下则是控制一定量线程并发执行,拥有这些资源的线程释放部分资源就可以唤醒后继节点
tryReleaseShared同样是抛出异常方式,需要实现类去实现的

1.doReleaseShared

/**
 * 共享模式下的释放动作 -- 唤醒后继和确保传递
 * 说明: 在独占模式下,如果Signal则释放仅仅是调用unparkSuccessor
 */
private void doReleaseShared() {
	for (;;) {
		Node h = head;
		if (h != null && h != tail) {
			int ws = h.waitStatus;
			if (ws == Node.SIGNAL) {
				if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
					continue;            // loop to recheck cases
				unparkSuccessor(h);
			}
			else if (ws == 0 &&
					 !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
				continue;                // loop on failed CAS
		}
		if (h == head)                   // loop if head changed
			break;
	}
}