java

关注公众号 jb51net

关闭
首页 > 软件编程 > java > JUC 并发核心 AQS 同步队列

Java并发编程之JUC并发核心AQS同步队列原理剖析

作者：没头脑遇到不高兴

AbstractQueuedSynchronizer 简称 AQS，可能我们几乎不会直接去使用它，但它却是 JUC 的核心基础组件，支撑着 java 锁和同步器的实现，大神 Doug Lea 在设计 JUC 包时希望能够抽象一个基础且通用的组件以支撑上层模块的实现，AQS 应运而生

一、AQS介绍

队列同步器AbstractQueuedSynchronizer（简称AQS），AQS定义了一套多线程访问共享资源的同步器框架，是用来构建锁或者其他同步组件的基础框架，是一个依赖状态(state)的同步器。Java并发编程的核心在java.util.concurrent（简称juc）包，而juc包的大部分工具都是以AQS为基础进行构建的，例如Semaphore、ReentranLock、CountDownLatch、CyclicBarrier等，它的作者是鼎鼎大名的Doug Lea。

AQS具备特性

它维护了一个volatile int state（代表共享资源）和一个FIFO线程等待队列（多线程争用资源被阻塞时会进入此队列）。state的访问方式有三种:

AQS有两种资源共享方式：Exclusive（独占式） 和 Share（共享式）。所谓独占式是指依据AQS中的state控制状态，只有一个线程能够进行工作（其它参与调度的线程会进入阻塞状态，如ReentrantLock）；共享式是指，依据AQS中的state控制状态，可以有多个满足条件的线程同时执行（如Semaphore/CountDownLatch）。　

AQS定义两种队列

不同的自定义同步器争用共享资源的方式也不同。自定义同步器在实现时只需要实现共享资源state的获取与释放方式即可，至于具体线程等待队列的维护（如获取资源失败入队/唤醒出队等），AQS已经在顶层实现好了。一般通过定义内部类Sync继承AQS将同步器所有调用都映射到Sync对应的方法。

自定义同步器实现时主要实现以下几种方法：

实现自定义同步组件时，将会调用同步器提供的模板方法，这些（部分）模板方法与描述如下。同步器提供的模板方法基本上分为3类：独占式获取与释放同步状态、共享式获取与释放同步状态和查询同步队列中的等待线程情况。自定义同步组件将使用同步器提供的模板方法来实现自己的同步语义。

二、AQS中的队列

1、同步等待队列

AQS当中的同步等待队列也称CLH队列，CLH队列是Craig、Landin、Hagersten三人发明的一种基于双向链表数据结构的队列，是FIFO先入先出线程等待队列，Java中的CLH队列是原CLH队列的一个变种，线程由原自旋机制改为阻塞机制。

这种结构的特点是每个数据结构都有两个指针，分别指向直接的后继节点和直接前驱节点。所以双向链表可以从任意一个节点开始很方便的访问前驱和后继。每个 Node 其实是由线程封装，当线程争抢锁失败后会封装成 Node 加入到 ASQ 队列中去；当获取锁的线程释放锁以后，会从队列中唤醒一个阻塞的节点线程 。

2、条件等待队列

条件等待队列是单向链表实现的，此时Node（下面会介绍）中pre和next都为null。Condition是一个多线程间协调通信的工具类，使得某个、或者某些线程一起等待某个条件（Condition），只有当该条件具备时，这些等待线程才会被唤醒，从而重新争夺锁。

3、AQS队列节点Node

同步队列中的节点（Node）用来保存获取同步状态失败的线程引用、等待状态以及前驱和后继节点，节点的属性类型与名称等，Node类基本属性定义如下所示，它是在AbstractQueuedSynchronizer中的一个内部类。

注意：如果Node在条件队列当中，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;
	volatile int waitStatus;
	volatile Node prev;
	volatile Node next;
	volatile Thread thread;
	Node nextWaiter;
}

Node pre：前驱节点，当前节点加入到同步队列中被设置（尾部添加）

Node next：后继节点

Thread thread：节点同步状态的线程

Node nextWaiter：等待队列中的后继节点，如果当前节点是共享的，那么这个字段是一个SHARED常量，也就是说节点类型(独占和共享)和等待队列中的后继节点共用同一个字段

int waitStatus：等待状态，标记当前节点的信号量状态 (1,0,-1,-2,-3)5种状态，使用CAS更改状态，volatile保证线程可见性，高并发场景下，即被一个线程修改后，状态会立马让其他线程可见，五种状态分别为：

三、同步队列源码分析

1、同步队列分析

同步器拥有首节点（head）和尾节点（tail），没有成功获取同步状态的线程将会成为节点加入该队列的尾部，同步队列的基本结构如图所示。

同步器包含了两个节点类型的引用，一个指向头节点，而另一个指向尾节点。同步器提供了一个基于CAS的设置尾节点的方法：compareAndSetTail(Node expect,Node update)，它需要传递当前线程“认为”的尾节点和当前节点，只有设置成功后，当前节点才正式与之前的尾节点建立关联。 涉及两个变化：

head节点表示获取锁成功的节点，当头结点在释放同步状态时，会唤醒后继节点，如果后继节点获得锁成功，会把自己设置为头结点，节点的变化过程如下

同步队列遵循FIFO，首节点是获取同步状态成功的节点，首节点的线程在释放同步状态时，将会唤醒后继节点，而后继节点将会在获取同步状态成功时将自己设置为首节点，该过程如下图所示。涉及两个变化：

设置首节点是通过获取同步状态成功的线程来完成的，由于只有一个线程能够成功获取到同步状态，因此设置头节点的方法并不需要使用CAS来保证，它只需要将首节
点设置成为原首节点的后继节点并断开原首节点的next引用即可。

2、同步队列——独占模式源码分析

acquire方法（独占获取）源码分析

通过调用同步器的acquire(int arg)方法可以获取同步状态，该方法对中断不敏感，也就是由于线程获取同步状态失败后进入同步队列中，后续对线程进行中断操作时，线程不会从同步队列中移出。

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

上面方法中首先调用自定义同步器实现的tryAcquire(int arg)方法（重写该方法），该方法保证线程安全的获取同步状态，如果同步状态获取失败，则构造同步节点（独占式Node.EXCLUSIVE，同一时刻只能有一个线程成功获取同步状态）并通过addWaiter(Node node)方法将该节点加入到同步队列的尾部，最后调用acquireQueued(Node node,int arg)方法，使得该节点以“死循环”的方式获取同步状态。如果获取不到则阻塞节点中的线程，而被阻塞线程的唤醒主要依靠前驱节点的出队或阻塞线程被中断来实现。

下面看下addWaiter方法的实现：把当前线程构建为Node节点；判断尾结点是否为空，通过CAS的方式将当前节点放到队列尾部；如果尾结点不为空或者前面CAS插入尾结点失败，调用enq方法，通过自旋的方式插入尾结点。

private Node addWaiter(Node mode) {
	// 1. 将当前线程构建成Node类型
	Node node = new Node(Thread.currentThread(), mode);
	// Try the fast path of enq; backup to full enq on failure
	Node pred = tail;
	// 2. 1当前尾节点是否为null？
	if (pred != null) {
		// 2.2 将当前节点尾插入的方式
		node.prev = pred;
		// 2.3 CAS将节点插入同步队列的尾部
		if (compareAndSetTail(pred, node)) {
			pred.next = node;
			return node;
		}
	}
	enq(node);
	return node;
}

下面看下enq方法的实现：判断尾结点是否为空，如果为空，则通过CAS的方式创建一个空的头结点（Thread为空），并将尾结点也指向头结点；如果尾结点不为空或者上面CAS创建头结点失败，将当前队列的前驱指针指向原来的尾结点，通过CAS的方式将当前节点放到队列尾部，将原来尾结点的后继指针指向当前节点；如果前面都失败了，进行下一次循环。当前线程构造的node节点通过addWaiter方法执行入队之后，其waitStatus为0，头结点的waitStatus也是0，此时是下面这种结构

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;
			//set尾部节点
			if (compareAndSetTail(t, node)) {//当前节点置为尾部
				t.next = node; //前驱节点的next指针指向当前节点
				return t;
			}
		}
	}
}

通过addWaiter 方法把线程添加到链表后， 会接着把 Node 作为参数传递给acquireQueued 方法，去竞争锁

/**
 * 已经在队列当中的Thread节点，准备阻塞等待获取锁
 */
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)) {//如果前驱结点是头结点，才tryAcquire，其他结点是没有机会tryAcquire的。
				setHead(node);//获取同步状态成功，将当前结点设置为头结点。
				p.next = null; // help GC
				failed = false;
				return interrupted;
			}
			/**
			 * 如果前驱节点不是Head，通过shouldParkAfterFailedAcquire判断是否应该阻塞
			 * 前驱节点信号量为-1，当前线程可以安全被parkAndCheckInterrupt用来阻塞线程
			 */
			if (shouldParkAfterFailedAcquire(p, node) &&
					parkAndCheckInterrupt())
				interrupted = true;
		}
	} finally {
		if (failed)
			cancelAcquire(node);
	}
}

下面是setHead方法的实现

private void setHead(Node node) {
        head = node;
        node.thread = null;
        node.prev = null;
    }

在acquireQueued(final Node node,int arg)方法中，当前线程在“死循环”中尝试获取同步状态，而只有前驱节点是头节点才能够尝试获取同步状态，这是为什么？原因有两个，如下。

如果前驱节点不是Head，通过shouldParkAfterFailedAcquire判断是否应该阻塞：如果前驱节点waitStatus为-1（SIGNAL的状态），当前线程可以安全的被parkAndCheckInterrupt用来阻塞线程；通过循环扫描链表把 CANCELLED 状态的节点移除；如果前驱节点waitStatus不是-1，则通过CAS将前驱节点的waitStatus改为-1。

第一次循环进入shouldParkAfterFailedAcquire方法时head节点为0，会将其改为SIGNAL，此时会返回false，那么外层的方法acquireQueued方法会执行第二次循环进入shouldParkAfterFailedAcquire方法，此时会返回true，当前线程可以被阻塞，则调用parkAndCheckInterrupt（）方法阻塞当前线程，其底层调用的是UnSafe类里面的park方法。

shouldParkAfterFailedAcquire方法会将前驱节点的waitStatus改为SIGNAL，因为只有前驱节点的状态是SIGNAL后继节点才可以被阻塞，次数除了tail节点的状态是0，其他的都是-1。

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
	int ws = pred.waitStatus;
	if (ws == Node.SIGNAL)
		/*
		 * 若前驱结点的状态是SIGNAL，意味着当前结点可以被安全地park
		 */
		return true;
	if (ws > 0) {
		/*
		 * 前驱节点状态如果被取消状态，将被移除出队列
		 */
		do {
			node.prev = pred = pred.prev;
		} while (pred.waitStatus > 0);
		pred.next = node;
	} else {
		/*
		 * 当前驱节点waitStatus为 0 or PROPAGATE状态时
		 * 将其设置为SIGNAL状态，然后当前结点才可以可以被安全地park
		 */
		compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
	}
	return false;
}

通过分析acquireQueued方法可以得出结论：头结点是获取同步状态成功的节点，头结点的所有有效后继节点线程都会被阻塞，释放锁后需要挨个唤醒头结点的后续线程节点。

独占式同步状态获取流程，也就是acquire(int arg)方法调用流程，如图所示。

release方法（独占释放）源码分析

当前线程获取同步状态并执行了相应逻辑之后，就需要释放同步状态，使得后续节点能够继续获取同步状态。通过调用同步器的release(int arg)方法可以释放同步状态，该方法在tryRelease方法释放了同步状态之后，会唤醒其后继节点（进而使后继节点重新尝试获取同步状态）。

public final boolean release(int arg) {
	if (tryRelease(arg)) {//释放一次锁
		Node h = head;
		if (h != null && h.waitStatus != 0)
			unparkSuccessor(h);//唤醒后继结点
		return true;
	}
	return false;
}

该方法执行时，会唤醒头节点的后继节点线程，unparkSuccessor(Node node)方法底层使用UnSafe的unpark方法来唤醒处于等待状态的线程。

private void unparkSuccessor(Node node) {
	//获取wait状态
	int ws = node.waitStatus;
	if (ws < 0)
		compareAndSetWaitStatus(node, ws, 0);// 将等待状态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);//唤醒线程
}

上面方法中判断node的后继节点是空或者waitStatus是撤销状态，会从tail往前遍历找到一个离node节点最近的节点，这是为什么呢？ 原因在于上面acquire时调用的enq入队方法：先compareAndSetTail(t, node)设置尾结点，然后t.next=node将前驱节点的next指针指向当前节点，如果t.next=node还没有执行的话，链表还没有建立完整，从前向后遍历时会出现遍历到t时找不到t的后继节点，从后往前遍历则不会出现这种情况。

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;
			//set尾部节点
			if (compareAndSetTail(t, node)) {//当前节点置为尾部
				t.next = node; //前驱节点的next指针指向当前节点
				return t;
			}
		}
	}
}

分析了独占式同步状态获取和释放过程后，适当做个总结：在获取同步状态时，同步器维护一个同步队列，获取状态失败的线程都会被加入到队列中（都是头结点的后继节点）并在队列中进行自旋；移出队列（或停止自旋）的条件是前驱节点为头节点且成功获取了同步状态。在释放同步状态时，同步器调用tryRelease(int arg)方法释放同步状态，然后唤醒头节点的后继节点。

3、同步队列——共享模式源码分析

同步队列共享模式与独占模式异同点：

acquireShared方法（共享获取）源码分析

acquireShared方法会先调用tryAcquireShared获取同步状态，如果返回值小于0表示获取失败，需要进行排队；如果获取成功，则可以向下执行。

public final void acquireShared(int arg) {
	if (tryAcquireShared(arg) < 0)//返回值小于0，获取同步状态失败，排队去；获取同步状态成功，直接返回去干自己的事儿。
		doAcquireShared(arg);
}

获取失败时会调用doAcquireShared方法进行排队：

public final boolean releaseShared(int arg) {
	if (tryReleaseShared(arg)) {
		doReleaseShared();
		return true;
	}
	return false;
}

到此这篇关于Java并发编程之JUC并发核心AQS同步队列原理剖析的文章就介绍到这了,更多相关JUC并发核心AQS同步队列内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:
阅读全文