并发和多线程(十五)--AbstractQueuedSynchronizer共享锁和Condition条件队列

Posted 2022-01-09

目录

• acquireShared()获取共享锁：
• doAcquireShared()
• setHeadAndPropagate()：
• 条件队列：
• signal()
• doSignal()
• transferForSignal()

在上篇博客中了解了排他锁的基本源码实现，所以现在我们学习下共享锁的源码，二者的源码实现大部分都是相同的，而且了解了排他锁的原理之后，我们现在阅读共享锁的源码会更加得心应手。

排他锁：当前锁只能被一个线程所持有，也只能有一个线程释放。

共享锁：当前锁可以被多个线程持有，并且可以设置持有锁的线程数量。

acquireShared()获取共享锁：

//共享锁获取锁
public final void acquireShared(int arg) 
	if (tryAcquireShared(arg) < 0)
		doAcquireShared(arg);

//排他锁获取锁
public final void acquire(int arg) 
	if (!tryAcquire(arg) &&
		acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
		selfInterrupt();

从上面代码看到共享锁尝试获取锁的方法不同了，而且返回值不再是boolean，而是int类型，大于0和小于0分别表示当前是否有线程持有锁。方法需要子类实现，我们在后面去学习lock相关类的时候再去了解。

doAcquireShared()

private void doAcquireShared(int arg) 
	//生成共享锁节点调用addWaiter，就是把节点添加到同步队列尾部
	final Node node = addWaiter(Node.SHARED);
	boolean failed = true;
	try 
		boolean interrupted = false;
		for (;;) 
			final Node p = node.predecessor();
			if (p == head) 
				int r = tryAcquireShared(arg);
				//①
				if (r >= 0) 
					//②共享锁，排他锁的主要区别在这里
					setHeadAndPropagate(node, r);
					p.next = null; // help GC
					if (interrupted)
						selfInterrupt();
					failed = false;
					return;
				
			
			if (shouldParkAfterFailedAcquire(p, node) &&
				parkAndCheckInterrupt())
				interrupted = true;
		
	 finally 
		if (failed)
			cancelAcquire(node);
	

方法和排他锁的实现差不多，

①.如果p为head，也就是当前节点为head的后继节点，就会尝试获取锁，r>=0意味着当前有线程获取锁

②.排他锁，将node设置为head，而这里调用了setHeadAndPropagate()

setHeadAndPropagate()：

private void setHeadAndPropagate(Node node, int propagate) 
    //这里和排他锁相同
	Node h = head; // Record old head for check below
	setHead(node);
    
	if (propagate > 0 || h == null || h.waitStatus < 0 ||
		(h = head) == null || h.waitStatus < 0) 
		Node s = node.next;
		if (s == null || s.isShared())
			doReleaseShared();
	

6-11行的逻辑是，获取锁的节点被设置为head之后，判断后续节点是否是shared节点，通过nextWaiter判断(前面说过，nextWaiter对于同步队列来说就是判断共享锁和排他锁，而对于条件队列指的是下一个节点)，如果是，将其唤醒。

private void doReleaseShared() 

	for (;;) 
		Node h = head;
		//当前队列至少两个节点
		if (h != null && h != tail) 
			int ws = h.waitStatus;
			//如果head的状态的signal，后续节点需要唤醒
			if (ws == Node.SIGNAL) 
			
				if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
					continue;            // loop to recheck cases
				//唤醒后驱节点
				unparkSuccessor(h);
			
			//如果状态为0，不能被设置为可传播的，跳过
			else if (ws == 0 &&
					 !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
				continue;                // loop on failed CAS
		
		//如果head发生变化，就继续循环，直到不发生变化，break，因为这个方法获取锁和解锁都会调用，head可能发生变化
		if (h == head)                   // loop if head changed
			break;
	

总结：
共享锁和排他锁主要区别就是，当一个线程获取锁之后，会尝试唤醒其后面的节点，能够让多个线程同时拥有锁，例如读锁。

条件队列：

一直学习到目前，好像发现条件队列的存在没有必要一样，但是其实不是，下面一起来了解下。Condition存在的原因是：同步队列无法解决所有的使用场景，例如锁+队列的使用场景，同步队列决定线程获取锁，如果需要排队阻塞就要用到Condition，无论是线程池ThreadPoolExecutor还是LinkedBlockingQueue等，都用到了条件队列。条件队列对这些线程进行管理，通过await()和signal()阻塞和唤醒。
关于condition的实现在刚开始学习AQS时就了解过了，和object的wait()和notify、notifyAll思想一致。

同步队列：负责加锁，解锁，互斥访问，双向队列。
条件队列：负责同步协作，单向队列。

public final void await() throws InterruptedException 
    //响应中断，抛出异常
    if (Thread.interrupted())
        throw new InterruptedException();
    //①添加当前线程对应的节点到Condition queue尾部
    Node node = addConditionWaiter();
    //②释放节点，保证在加入条件队列阻塞之前，释放获取独占锁时的资源
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    //③当前节点如果不在同步队列，将自己挂起。
    while (!isOnSyncQueue(node)) 
        LockSupport.park(this);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    
    //从while break说明线程被signal，转移到同步队列，所以直接acquireQueued在同步队列阻塞，而且说明条件队列只能和独占锁结合使用
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    //代码执行到这里说明，已经获取到独占锁，删除cancel的节点
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);

①addConditionWaiter()

private Node addConditionWaiter() 
    //条件队列tail
    Node t = lastWaiter;
    // 如果tail状态不是condition，清楚条件队列中所有状态非condition的节点
    if (t != null && t.waitStatus != Node.CONDITION) 
        //就是遍历所有节点，剔除不符合的节点， 其中trail表示上一个状态
        unlinkCancelledWaiters();
        t = lastWaiter;
    
    //创建状态为condition的新节点
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    //如果条件队列为空，node为head，否则挂到后面
    if (t == null)
        firstWaiter = node;
    else
        t.nextWaiter = node;
    lastWaiter = node;
    return node;

②fullyRelease

final int fullyRelease(Node node) 
    boolean failed = true;
    try 
        int savedState = getState();
        //释放锁，并唤醒后续节点
        if (release(savedState)) 
            failed = false;
            return savedState;
         else 
            throw new IllegalMonitorStateException();
        
     finally 
        //如果失败，将节点状态设置为cancelled
        if (failed)
            node.waitStatus = Node.CANCELLED;
    

③isOnSyncQueue(node)：

当前节点此时不在同步队列的可能大概是有两种：

1.刚加入条件队列，然后就被其他线程signal转移到同步队列。

2.之前就在这阻塞，被唤醒加入同步队列。

final boolean isOnSyncQueue(Node node) 
    //如果节点waitStatus为CONDITION，或者prev为null，返回false
    if (node.waitStatus == Node.CONDITION || node.prev == null)
        return false;
    if (node.next != null) // If has successor, it must be on queue，因为条件队列中没有next，而是nextWaiter
        return true;
    //从tail开始遍历查询节点
    return findNodeFromTail(node);

signal()

条件队列的阻塞通过signal或者signalAll()进行唤醒，和notify/notifyAll理念相同，下面来了解一下。

public final void signal() 
    //判断当前线程和持有锁的线程是否一致，不一致，直接抛出异常
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    //条件队列首节点不为null，doSignal
    Node first = firstWaiter;
    if (first != null)
        //将条件队列头结点转移到同步队列中去
        doSignal(first);

doSignal()

private void doSignal(Node first) 
    do 
        //当前遍历到队尾了
        if ( (firstWaiter = first.nextWaiter) == null)
            
            lastWaiter = null;
        //从头结点开始遍历，这步操作就是把first从条件队列剔除
        first.nextWaiter = null;
        //transferForSignal把节点转移到同步队列中，(first = firstWaiter) != null为false表示遍历结束
     while (!transferForSignal(first) &&
             (first = firstWaiter) != null);

transferForSignal()

final boolean transferForSignal(Node node) 
    //CAS把node状态从condition设置为0,0就是初始化，失败返回false
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;
    //把当前节点添加到同步队列尾部，返回其前一个节点
    Node p = enq(node);
    int ws = p.waitStatus;
    //如果p被取消，或者状态设置为signal失败，都会唤醒当前线程，成功返回true
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);
    return true;

如果转移成功，返回true，否则返回false。这个过程和排他锁的获取锁的过程相似，将节点添加到队列尾部，将前驱节点状态设置signal。关于signalAll()方法这里就不写了，可以自行了解一下，代码差不多，多了个while循环。