zoukankan      html  css  js  c++  java
  • 深入理解Java并发类——AQS

    什么是AQS

    AbstractQueuedSynchronizer,是Java并发包中locks目录下的一个抽象类,它是实现各种同步结构和部分其他组成单元的基础,如线程池中的Worker

    为什么需要AQS

    Doug Lea曾经介绍过AQS的设计初衷。从原理上,一种同步结构往往是可以利用其他的结构实现的。但是,对某种同步结构的倾向,会导致复杂的实现逻辑,所以他选择将基础的同步相关操作抽象在AQS中,利用AQS为我们构建同步结构提供范本。
    抽象类的特点看这里

    AQS的核心思想

    如果被请求的共享资源空闲,则将请求资源的线程设为有效的工作线程,并将资源锁定,资源使用结束时释放锁定。如果资源被锁定,则把请求未果的线程加入FIFO队列,当资源可用时唤醒队头的线程。
    这个过程中,通过state判断资源是否锁定,并通过对state进行CAS操作,来蟹盖资源的锁定状态。

    AQS的内部数据和方法

    1. 一个volatile的整数成员表征状态,同时提供了setState、getState、CAS方法。

      private volatile int state;
      
      //当state的值与期望值相同时,把state设为update
      protected final boolean compareAndSetState(int expect, int update) {
           // See below for intrinsics setup to support this
           return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
       }     
      
    2. 一个先入先出(FIFO)的等待线程队列,以实现多线程间竞争和等待,这是AQS机制的核心之一

    3. 各种基于CAS的基础操作方法,以及各种期望具体同步结构去实现的acquire/release方法。

    如何利用AQS实现同步结构

    利用AQS实现一个同步结构,至少要实现两个基本类型的方法,分别是:

    1. acquire操作,获取资源的独占权.
    2. release操作,释放对某个资源的独占。

    AQS使用了模板设计模式,自定义同步容器要重写这些模板方法。

    //独占方式
    protected boolean tryAcquire(int arg) {
            throw new UnsupportedOperationException();
        }
    protected boolean tryRelease(int arg) {
            throw new UnsupportedOperationException();
        }
    //共享方式
    protected int tryAcquireShared(int arg) {
            throw new UnsupportedOperationException();
        }
    protected boolean tryReleaseShared(int arg) {
            throw new UnsupportedOperationException();
        }
    //该线程是否正在独占资源,只有用到condition时才重写它
    protected boolean isHeldExclusively() {
            throw new UnsupportedOperationException();
        }
    

    ReentrantLock对AQS的利用

    ReentrantLock为例,它内部通过继承AQS抽象类实现了Sync子类,以AQS的state来表示和操作锁的状态。

    private final Sync sync;
    absract static class Sync extends AbsractQueuedSynchronizer { 
        abstract void lock();
        final boolean nonfairTryAcquire(int acquires) {...}
        protected final boolean tryRelease(int releases) {...}
        ...
    }
    

    下面是ReentrantLock对应acquire和release操作,如果是CountDownLatch则可以看作是await()/countDown(),具体实现也有区别。
    初始化时staate=0,表示资源未锁定,线程A调用lock()后,state+1,此后其他线程再lock()就会失败,直到A进行unlock(),state-1。
    另外,A可以重复获取此锁,每次获取state累加,获取多少次就释放多少次,这就是可重入

    public void lock() {
     sync.acquire(1);
    }
    public void unlock() {
     sync.release(1);
    }
    

    尝试获取锁

    下面是线程试图获取锁的过程,acquire()的实现逻辑是在AQS的内部,调用了tryAcquire和acquireQueued。

    tryAcquire是按照特定场景需要开发者去实现的部分,而线程间竞争则是AQS通过Waiter队列与acquireQueued提供的。

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

    在ReentrantLock中,tryAcquire的逻辑实现在NonfairSync和FairSync中,分别提供了进一步的非公平或公平性方法,而AQS内部tryAcquire仅仅是个接近未实现的方法(直接抛异常)。

    ReentrantLock默认是非公平的,下面是公平性在ReentrantLock构建时如何设定的。

    public ReentrantLock() {
     sync = new NonfairSync(); // 默认是非公平的
     }
     public ReentrantLock(boolean fair) {
     sync = fair ? new FairSync() : new NonfairSync();
     }
    

    以非公平的tryAcquire为例,其内部实现了如何配合状态与CAS获取锁。对比公平版本的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()) { //虽然状态不是0,但当前线程是锁持有者,则进行重入
             int nextc = c + acquires;
             if (nextc < 0) // overflow
                 throw new Error("Maximum lock count exceeded");
             setState(nextc);
             return true;
         }
         return false;
    }
    

    获取锁失败,排队竞争

    接下来分析acquireQueued,如果前面的tryAcquire失败,代表着锁争抢失败,进入排队竞争阶段

    当前线程会被包装成为一个排他模式的节点(EXCLUSIVE),通过addWaiter方法添加到FIFO队列中。

    如果当前节点的前面是头节点,则试图获取锁,一切顺利则成为新的头节点;否则,有必要则等待。

    final boolean acquireQueued(final Node node, int arg) {
     boolean interrupted = false;
     try {
     for (;;) {// 循环
     final Node p = node.predecessor();// 获取前一个节点
     if (p == head && tryAcquire(arg)) { // 如果前一个节点是头结点,表示当前节点合适去tryAcquire
     setHead(node); // acquire成功,则设置新的头节点
     p.next = null; // 将前面节点对当前节点的引用清空
     return interrupted;
     }
     if (shouldParkAfterFailedAcquire(p, node)) // 检查是否失败后需要park
     interrupted |= parkAndCheckInterrupt();
     }
     } catch (Throwable t) {
     cancelAcquire(node);// 出现异常,取消
     if (interrupted)
     selfInterrupt();
     throw t;
     }
    }
    

    总结。以上是对ReentrantLock的acquire方法的分析,release方法与之相似。

    参考

    《Java并发编程实战》
    《Java核心技术36讲》杨晓峰

  • 相关阅读:
    XGBoost算法--学习笔记
    机器学习--学习书籍
    一天搞懂深度学习-深度学习新浪潮
    如何在 Office 365 环境中设置联机 Exchange 邮箱大小和限制
    玩转Office 365中的Exchange Online服务 之十一 怎样在Exchange Online中配置邮件传递限制《转》
    玩转Office 365中的Exchange Online服务 之六 了解Exchange Online对于邮箱使用的限制《转》
    Hyper-V Ubuntu修改分辨率
    k8s 集群基本概念<转>
    Azure 中 Linux VM 的 SSH 公钥和私钥对
    docker学习笔记(k8s) 《转》
  • 原文地址:https://www.cnblogs.com/ChengzhiYang/p/12487155.html
Copyright © 2011-2022 走看看