[Java知识库]手把手讲解AQS源码

手把手讲解AQS源码

一、概述

? 本文将会通过ReentrantLock为例，带大家看一下AQS的源码，其实并不难，下面是一个公平锁的小案例，大家可以自己跑一下感受一下。下面将会带大家一点一点阅读源码，认真看下来你就会发现其实并不难。

/**
 * @author VernHe
 * @date 2021年12月02日
 */
public class TestAQS {
    /**
     * true表示公平锁，false表示非公平锁
     */
    static ReentrantLock reentrantLock = new ReentrantLock(true);

    public static void main(String[] args) {
        for (int i = 0; i < 5; i++) {
            new Thread(new Target(i)).start();
        }
    }

    static class Target implements Runnable {
        // 每个任务的编号
        private int num;

        public Target(int num) {
            this.num = num;
        }

        @Override
        public void run() {
            for (int i = 0; i < 2; i++) {
                reentrantLock.lock();
                try {
                    System.out.println("任务" + this.num + "执行了");
                } finally {
                	// unlock方法必须写在finally里面
                    reentrantLock.unlock();
                }
            }
        }
    }
}

二、源码部分

ReentrantLock

按住Ctrl+鼠标左键，点击lock()方法，就会进入到方法内部

public void lock() {
	sync.lock();
}

到这里，你会发现用的是sync的lock()方法，按住Ctrl点击sync就会发现它其实就是一个抽象静态内部类

public class ReentrantLock implements Lock, java.io.Serializable {


    private final Sync sync;

    /**
     * Base of synchronization control for this lock. Subclassed
     * into fair and nonfair versions below. Uses AQS state to
     * represent the number of holds on the lock.
     */
    abstract static class Sync extends AbstractQueuedSynchronizer {
    	//此处省略
    }
}

通过英文注释（看不懂的可以翻译看一下），可以看出，ReentrantLock是基于Sync类来实现公平/非公平锁，使用AQS中的state属性去表示加锁的次数，其实我们常说的AQS其实就是AbstractQueuedSynchronizer，继续点lock()方法往下走，选择FairSync，可以看到下面的源码

static final class FairSync extends Sync {

	final void lock() {
		acquire(1);
	}
	// ....省略
}

进入到acquire(1)中

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

这里解释一下if中的三个方法：

• tryAcquire()

  顾名思义，就是会尝试获取一次锁。公平/不公平锁的逻辑会有一点点的不同

  protected final boolean tryAcquire(int acquires) {
  	final Thread current = Thread.currentThread();
      // 获取state的值
  	int c = getState();
  	if (c == 0) {	// 如果state为0说明目前每人使用
  		if (!hasQueuedPredecessors() &&	// 公平：会判断它前面有没有其他线程，非公平则不会
  			compareAndSetState(0, acquires)) {	// CAS操作，尝试进行获取
               setExclusiveOwnerThread(current);	// 把自己设置成独占的线程
               return true;
            }
       }
       else if (current == getExclusiveOwnerThread()) {	// 如果有人使用并且使用的人是自己
  		int nextc = c + acquires;	// 每lock()一次就会让state的值增加1
           if (nextc < 0)
               throw new Error("Maximum lock count exceeded");
           setState(nextc);			// 更新state
           return true;
        }
        return false;
  }
  

  总结一下上面的方法其实很简单，在【没人使用并且当前线程成功独占】或者【正在独占的线程是自己】的时候才会返回true，说得通俗一点就是试一下看能不能独占

• addWaiter()

  翻译过来就是添加等待者，其实说白了，就是上面尝试获取锁失败后会加入到等待队列中

  private Node addWaiter(Node mode) {
  	Node node = new Node(Thread.currentThread(), mode);	// 为当前线程创建一个节点并设置相应的模式
  	// Try the fast path of enq; backup to full enq on failure
  	Node pred = tail; // 指向等待队列最后面的Node
  	if (pred != null) {	// 如果有线程也在等待
          node.prev = pred;
          if (compareAndSetTail(pred, node)) {
              pred.next = node; // 排到最后一个Node的后面
              return node; // 排队成功后返回当前线程对应的Node
          }
  	}
      enq(node); // 如果本线程是第一个排队的，或者前面排队没成功，则再次尝试排队直至成功为止
      return node; // 排队成功后返回当前线程对应的Node
  }
  

  总结一下，说得通俗一点就是排队

• acquireQueued()

  在这里面做CAS自旋，会不断获取锁，失败后会阻塞当前线程

  final boolean acquireQueued(final Node node, int arg) {
  	boolean failed = true;	// 记录是否独占失败
  	try {
          boolean interrupted = false; // 记录线程是否被打断
          for (;;) {	// 循环，直至成功独占
              final Node p = node.predecessor(); // 获取前一个Node
              if (p == head && tryAcquire(arg)) {	// 如果自己是第二个并且成功独占
                  setHead(node);	// 把当前Node设置成新的head
                  p.next = null; // help GC
                  failed = false;
                  return interrupted;	// 返回，跳出循环
              }
              if (shouldParkAfterFailedAcquire(p, node) && //检查自己是否应该阻塞
                  parkAndCheckInterrupt()) // 阻塞当前线程（debug会停在这一步）
                  interrupted = true;	 // 当线程被重新唤醒时才会知心这个方法，然后继续循环
  		}
  	} finally {
  		if (failed)
  			cancelAcquire(node); // 如果独占失败，则会唤醒后面的Node继续执行此方法
  	}
  }
  

  ?

AbstractQueuedSynchronizer

组成部分

? 1、等待队列（CLH队列），其实本质上是双向链表

? 2、state变量，保存同步状态

? 3、head 和 tail指针，用于保存等待队列的头尾

? 4、内部类Node,通过两个指针，之前前驱节点和后继节点

? 5、操作队列的方法和一系列的CASnative方法

三、总结

? 个人理解，AQS框架抽取了很多同步策略的特性，比如信号量、互斥量、各种锁中都可能出现的有的线程需要等待的情况，为此，抽取出一个阻塞队列（CLH）用于保存阻塞的线程并用于之后唤醒。不同的同步策略所允许的同时运行的线程的数量不一样，为此，抽取出一个state变量。之后就是一系列的CAS方法来操作阻塞队列，并且底层都是C语言实现的原子操作。