专栏首页java、Spring、技术分享AbstractQueuedSynchronizer源码解读

AbstractQueuedSynchronizer源码解读

   AbstractQueuedSynchronizer(AQS),是 Java 并发包中,实现各种同步结构和部分其他组成单元(如线程池中的 Worker)的基础。AQS 内部数据和方法,可以简单拆分为:

  • 一个 volatile 的整数成员表征状态(state属性),同时提供了 setState 和 getState 方法。 1: ReentrantLock: 用state变量记录某个线程获取独占锁的次数,获取锁时+1,释放锁时-1,在获取时会校验线程是否可以获取锁。 2:Semaphore: 用state变量作为计数器,只有在大于0时允许线程进入。获取锁时-1,释放锁时+1。 3: CountDownLatch: 用state变量作为计数器,在初始化时指定。只要state还大于0,获取共享锁会因为失败而阻塞,直到计数器的值为0时,共享锁才允许获取,所有等待线程会被逐一唤醒。
  • 一个先入先出(FIFO)的等待线程队列,以实现多线程间竞争和等待,这是 AQS 机制的核心之一。
  • 各种基于 CAS 的基础操作方法,以及各种期望具体同步结构去实现的 acquire/release 方法。

AbstractQueuedSynchronizer关键属性

下面来看看acquire获取锁的部分源码:

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
//定义为final说明子类不能覆盖这个实现过程
    public final void acquire(int arg) {
//这里分别会调用tryAcquire方法用于获取锁,如果失败,会调用addWaiter方法,将当前线的Node加入到FIFO队列里,通过acquireQueued判断当前节点是否为头节点,如果是则试图获取锁,如果不是,则进行等待。
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

//这个方法在AbstractQueuedSynchronizer里留给子类来实现,
    protected boolean tryAcquire(int arg) {
        throw new UnsupportedOperationException();
    }

}

这里分别会调用tryAcquire方法用于获取锁,如果失败,会调用addWaiter方法,将当前线的Node加入到FIFO队列里,通过acquireQueued判断当前节点是否为头节点,如果是则试图获取锁,如果不是,则进行等待。在AbstractQueuedSynchronizer里tryAcquire方法是留给子类来实现的,下面来看看ReentrantLock里的非公平锁是如何实现这里的逻辑的。源码如下:

abstract static class Sync extends AbstractQueuedSynchronizer {
      //非公平锁最终最调用到当前这个方法,传入的acquires为1
    final boolean nonfairTryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
         int c = getState();// 获取当前 AQS 内部状态量
          if (c == 0) { // 0 表示无人占有,则直接用 CAS 修改状态位,
                if (compareAndSetState(0, acquires)) {// 不检查排队情况,直接争抢
                      setExclusiveOwnerThread(current);  // 并设置当前线程独占锁
                    return true;
                }
          } else if (current == getExclusiveOwnerThread()) { // 即使状态不是 0,也可能当前线程是锁持有者,因为这是再入锁
                int nextc = c + acquires;
                  if (nextc < 0) // overflow
                throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
    }
    return false;
}


}

ReentrantLock的公平锁的实现逻辑如下:

   static final class FairSync extends Sync {
//公平锁的实现逻辑
        protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();// 获取当前 AQS 内部状态量
            if (c == 0) {// 0 表示无人占有,
                if (!hasQueuedPredecessors() && //这里需要判断队列头节点的的下一个节点为空。也就是说不允许线程进行插队。
                    compareAndSetState(0, acquires)) { //为空才会尝试获取锁
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            else if (current == getExclusiveOwnerThread()) { // 即使状态不是 0,也可能当前线程是锁持有者,因为这是再入锁
                int nextc = c + acquires;
                if (nextc < 0)
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }
    }

接下来再分析 acquireQueued,如果前面的 tryAcquire 失败,代表着锁争抢失败,进入排队竞争阶段。这里就是利用 FIFO 队列,实现线程间对锁的竞争的部分,算是是 AQS 的核心逻辑。源码如下:

//这个方法是不允许子类重写的
    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)) { / 如果前一个节点是头结点,则当前节点是FIFO队列里的第一个节点,则会再次调用tryAcquire方法尝试获取锁。
                    setHead(node); //成功,则设置当前节点为头节点
                    p.next = null; // help GC 清空前面节点对当前节点的引用
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) && //失败后需要调用park
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed) // 获取锁失败,调用cancelAcquire方法
                cancelAcquire(node);
        }
    }

acquireQueued 的逻辑,简要来说,就是如果当前节点的前面是头节点,则试图获取锁,一切顺利则成为新的头节点;否则,有必要则等待。上面的部分分析了获取锁的逻辑。下面再来看看释放锁的逻辑AbstractQueuedSynchronizer的relase源码如下:

//释放锁需要调用release方法
    public final boolean release(int arg) {
        if (tryRelease(arg)) {
//释放成功后,调用unparkSuccessor方法,让下个节点得到锁
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        
//留给具体的子类来实现
    protected boolean tryRelease(int arg) {
        throw new UnsupportedOperationException();
    }

//当前节点释放锁,下个节点得到锁状态 
    private void unparkSuccessor(Node node) {
        int ws = node.waitStatus;
//设置当前节点的waitStatus
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);
//当前节点的下一个节点
        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;
        }
//下一个节点不为空,则调用LockSupport的unpark方法唤醒线程
        if (s != null)
            LockSupport.unpark(s.thread);
    }

ReentrantLock里tryRelease的实现如下所示:

        protected final boolean tryRelease(int releases) {
            int c = getState() - releases; //用state减去release, 这个状态是新的状态 
            if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
            boolean free = false; //释放的结果用free来表示
            if (c == 0) { //为零,表求当前没有钱程点用了,
                free = true;
                setExclusiveOwnerThread(null);
            }
  //设置状态 
            setState(c);
            return free;
        }

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 深入分析Spring MVC中RequestBody与ResponseBody

      在SpringMVC中,可以使用@RequestBody和@ResponseBody两个注解,分别完成请求报文到对象和对象到响应报文的转换。在Sprin...

    良辰美景TT
  • Netty NioEventLoop源码解读

      NioEventLoop中维护了一个线程,线程启动时会调用NioEventLoop的run方法,执行I/O任务和非I/O任务:I/O任务:即selectio...

    良辰美景TT
  • Netty 编码与数据的写入

    业务里的数据最终需要通过socket写回到客户端,我们写的业务方法都是面向对象来进行编码的,而channel底层传输的是字节,Netty通过定义encode...

    良辰美景TT
  • 高并发编程-CountDownLatch深入解析

    CountDownLatch允许一个或者多个线程一直等待,直到一组其它操作执行完成。在使用CountDownLatch时,需要指定一个整数值,此值是线程将要等待...

    JavaQ
  • Android 9.0 SystemUI 下拉状态栏快捷开关

    SystemUI 下拉状态栏快捷开关是 QSPanel,qs_panel.xml,@+id/quick_settings_panel,本篇文章就来看看这些快捷开...

    吴小龙同學
  • 深入浅出AQS之共享锁模式

    原文链接:https://www.jianshu.com/p/1161d33fc1d0

    天涯泪小武
  • TreeMap之红黑树【源码篇】

    前序: 在用TreeMap的时候,发现其非常有特点,故结合网上的资料整合写了此篇。 红黑树:红黑树并不是一个完美平衡二叉查找树,根结点P的左子树显然比右子树高,...

    奕仁
  • 分布式锁,三种实现方式,性能对比分析!

    日常工作中很多场景下需要用到分布式锁,例如:任务运行(多个节点同一时刻同一个任务只能在一个节点上运行(分片任务除外)),交易请求接收(前端交易请求发送时,可能由...

    搜云库技术团队
  • 从0到1搭建spark集群---企业集群搭建

    今天分享一篇从0到1搭建Spark集群的步骤,企业中大家亦可以参照次集群搭建自己的Spark集群。

    LhWorld哥陪你聊算法
  • 各种字符串问题

    总结:所有题目都已做,有些Easy没有做第二遍,有两道没有accept,请戳 link-en, link-cn

    王脸小

扫码关注云+社区

领取腾讯云代金券