热门关键字:
jquery > jquery教程 > java > AQS源码分析--jdk1.8

AQS源码分析--jdk1.8

364
作者:管理员
发布时间:2020/4/14 15:29:22
评论数:0
转载请自觉注明原文:http://www.jq-school.com/Show.aspx?id=1303

JDK1.8

ArrayList源码分析--jdk1.8
LinkedList源码分析--jdk1.8
HashMap源码分析--jdk1.8
AQS源码分析--jdk1.8
ReentrantLock源码分析--jdk1.8

AbstractQueuedSynchronizer概述

  1. AQS是一个基于FIFO队列,可以用于构建锁或者其他相关同步装置的基础框架。
  2. AQS提供了双向链表。
  3. AQS分为共享模式和独占模式。
  4.AQS基于volatile内存可见性和CAS原子性操作实现线程间通信操作。

AbstractQueuedSynchronizer数据结构

  数据结构是集合的精华所在,数据结构往往也限制了集合的作用和侧重点,了解各种数据结构是我们分析源码的必经之路。
  AQS的数据结构如下:双向链表
  AQS源码分析--jdk1.8
  AQS实现共享资源的访问控制基础:
     1.state字段,即同步器状态字段。用于共享资源的访问控制
     2.CLH队列,FIFO等待队列,存放竞争失败的线程。通常CLH队列是一个自旋队列,AQS以阻塞的方式实现
     CLH队列的使用:
AQS源码分析--jdk1.8

CLH扫盲

自旋锁
学习了解自旋锁之前先回顾一下互斥锁 
互斥锁 
线程在获取互斥锁的时候,如果发现锁已经被其它线程占有,那么线程就会惊醒休眠,然后在适当的时机(比如唤醒)在获取锁。 
自旋锁 
那么自旋锁顾名思义就是“自旋”。就是当一个线程在尝试获取锁失败之后,线程不会休眠或者挂起,而是一直在循环检测锁是否被其它线程释放。 
区别 
互斥锁就是开始开销要大于自旋锁。临界区持锁时间的大小并不会对互斥锁的开销造成影响,而自旋锁是死循环检测,加锁全程消耗cpu,起始开销虽然低于互斥锁,但是随着持锁时间,加锁的开销是线性增长。 
适用的情况 
互斥锁用于临界区持锁时间比较长的操作,比如下面这些情况都可以考虑

临界区有IO操作
临界区代码复杂或者循环量大
临界区竞争非常激烈
单核处理器 
自旋锁就主要用在临界区持锁时间非常短且CPU资源不紧张的情况下。当递归调用时有可能造成死锁。 
线程(节点)队列 
了解了自旋锁之后,在学习ReentrantLock的时候,一个线程在等待锁的时候会被封装成一个Node节点,然后加入一个队列中并检测前一个节点是否是头节点,并且尝试获取锁,如果获取锁成功就返回,否则就阻塞。直到上一个节点释放锁并唤醒它。这样看来似乎跟自旋没什么挂钩。这是因为AQS里面的CLH队列是CLH队列锁的一种变形。先来了解一下CLH队列锁
CLH队列锁 
CLH(Craig, Landin, and Hagersten locks): 是一个自旋锁,能确保无饥饿性,提供先来先服务的�%8心方法分析

1. acquire方法--独占模式

     1)acquire(int arg);
     以独占模式获取资源,如果获取成功,直接返回,否则进去CLH等待队列,通过自旋知道获取到资源为止,过程中忽略线程中断,获取资源后才进行自我中断(补上),下面看源码:

/**
 * AQS的独占模式--互斥
 * tryAcquire()尝试直接去获取资源,如果成功则直接返回;
 * addWaiter()将该线程加入等待队列的尾部,并标记为独占模式;
 * acquireQueued()使线程在等待队列中获取资源,一直获取到资源后才返回。如果在整个等待过程中被中断过,则返回true,否则返回false。
 * 如果线程在等待过程中被中断过,它是不响应的。只是获取资源后才再进行自我中断selfInterrupt(),将中断补上。
 */ public final void acquire(int arg) { if (!tryAcquire(arg) && // 再次尝试上锁 回到了  NonfairSync.tryAcquire 方法, tryAcquire 调用了 Sync.nonfairTryAcquire方法
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) // 链表尾部添加节点 为独占模式
        selfInterrupt();
}

     2)boolean tryAcquire(int arg);
     尝试以独占的方式获取资源,成功true,失败false,该方法可以用于实现Lock中的tryLock()方法。

/**
 * tryAcquire尝试以独占的方式获取资源,如果获取成功,则直接返回true,否则直接返回false。该方法可以用于实现Lock中的tryLock()方法。
 * 该方法的默认实现是抛出UnsupportedOperationException,具体实现由自定义的扩展了AQS的同步类来实现。AQS在这里只负责定义了一个公共的方法框架。
 * 这里之所以没有定义成abstract,是因为独占模式下只用实现tryAcquire-tryRelease,而共享模式下只用实现tryAcquireShared-tryReleaseShared。
 * 如果都定义成abstract,那么每个模式也要去实现另一模式下的接口
 * 由子类选择性实现
 */ protected boolean tryAcquire(int arg) { throw new UnsupportedOperationException();
}

     3)Node addWaiter(Node mode);
     将一个Node节点放入到CLH队列的队尾。

/**
 * 将一个Node节点放入到CLH队列的队尾。
 * 第一步:首先将oldTail赋值给newNode.prev:node.prev = pred, 把当前tail节点赋值到mode新节点的prev前一个,
 * 第二步:将tail赋值给newNode:compareAndSetTail(pred, node) 把当前tail节点的内存地址修改为(指向)新的mode节点,
 * 第三步:将oldTail的next指针指向newNode(即tail):pred.next = node 把当前tail节点的next后一个赋值为新的mode节点(即tail)
 * 如果队列为空,通过enq(node)方法初始化一个等待队列,并返回当前节点
 */ private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode); // Try the fast path of enq; backup to full enq on failure //尝试快速入队,失败则使用enq()方式
    Node pred = tail; if (pred != null) { // 列队尾部不为空
        node.prev = pred; if (compareAndSetTail(pred, node)) {
            pred.next = node; return node;
        }
    } // 列队尾部为空 或者  CAS 操作失败
    enq(node); return node;
}

     4)boolean acquireQueued(final Node node, int arg);
     使线程在等待队列中获取资源,一直获取到资源后才返回。如果在整个等待过程中被中断过,则返回true,否则返回false。

/**
 * 若node节点的前继节点是head节点,则会再次调用tryAcquire()获取资源
 * 判断当前节点的前继节点是否为head节点。若是,则表示该节点有资格尝试获取共享资源。此处的head节点的判断在一定程度上保证资源竞争的公平性
 * shouldParkAfterFailedAcquire():判断当前节点是否可以安全进入park()
 * parkAndCheckInterrupt():让线程进入等待
 * 用于队列中的线程自旋地以独占且不可中断的方式获取同步状态(acquire),直到拿到锁之后再返回。该方法的实现分成两部分:
 * 如果当前节点已经成为头结点,尝试获取锁(tryAcquire)成功,然后返回;否则检查当前节点是否应该被park,然后将该线程park并且检查当前线程是否被可以被中断
 */ final boolean acquireQueued(final Node node, int arg) { //标记是否成功拿到资源,默认false boolean failed = true; try { boolean interrupted = false;//标记等待过程中是否被中断过 for (;;) { final Node p = node.predecessor(); // 判断当前节点的 前驱节点 是否为队列头部  如果是 再次尝试上锁(如果头部节点 已经释放锁, 则使当前线程成为持有者 并且设置自己为 头部。 同时释放前驱节点) if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false; return interrupted;
            } //判断当前节点是否可以进入park,若可以,让线程进入等待 if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally { //如果获取资源失败,则取消 if (failed)
            cancelAcquire(node);
    }
}

     5)void selfInterrupt();
     中断当前线程

/**
 * Convenience method to interrupt current thread.
 * 中断当前线程
 */ static void selfInterrupt() {
    Thread.currentThread().interrupt();
}

     6)Node enq(final Node node);
     将当前节点插入等待队列

/**
 * 进行自旋入队方式的enq()方法,基本和addWaiter()方法一致:
 * 用于将当前节点插入等待队列,如果队列为空,则初始化当前队列。整个过程以CAS自旋的方式进行,直到成功加入队尾为止
 */ private Node enq(final Node node) { for (;;) {
        Node t = tail; if (t == null) { // Must initialize 必须初始化 尾部为空 尝试构建表结构 if (compareAndSetHead(new Node()))
                tail = head;
        } else { //尾部不为空 不断尝试  CAS 操作
            node.prev = t; if (compareAndSetTail(t, node)) {
                t.next = node; return t;
            }
        }
    }
}

     7)boolean compareAndSetHead(Node update);
     通过原子(CAS)操作 改变上锁状态

/**
 * CAS head field. Used only by enq.
 * 通过原子操作 改变上锁状态
 * this == null
 * 第一个参数为需要改变的对象,第二个为偏移量(即之前求出来的valueOffset的值),第三个参数为期待的值,第四个为更新后的值
 */ private final boolean compareAndSetHead(Node update) { //调用本地方法 实现硬件级别的原子操作 cas return unsafe.compareAndSwapObject(this, headOffset, null, update);
}

     8)boolean parkAndCheckInterrupt();
     让线程去休息,真正进入等待状态

/**
 * 该方法让线程去休息,真正进入等待状态。park()会让当前线程进入waiting状态。在此状态下,有两种途径可以唤醒该线程:
 * 1)被unpark();2)被interrupt()。需要注意的是,Thread.interrupted()会清除当前线程的中断标记位
 */ private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this); // 又是一个底层类 实现线程等待 return Thread.interrupted();
}

     9)boolean shouldParkAfterFailedAcquire(Node pred, Node node);
     判断当前节点中的线程,是否可以安全的进入park()。返回true,表示进程可以进入park

/**
 * 该方法的作用在于判断当前节点中的线程,是否可以安全的进入park()。返回true,表示进程可以进入park。若前驱节点的waitStatus为SIGNAL,则表示当前节点�   failed = false; return;
                }
            } if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally { if (failed)
            cancelAcquire(node);
    }
}

     4)boolean parkAndCheckInterrupt();
     让线程去休息,真正进入等待状态

/**
 * 该方法让线程去休息,真正进入等待状态。park()会让当前线程进入waiting状态。在此状态下,有两种途径可以唤醒该线程:
 * 1)被unpark();2)被interrupt()。需要注意的是,Thread.interrupted()会清除当前线程的中断标记位
 */ private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this); // 又是一个底层类 实现线程等待 return Thread.interrupted();
}

     5)cancelAcquire(Node node);
     取消节点

/**
 * 取消节点
 * 列队等待中 抛出异常会调用此方法
 */ private void cancelAcquire(Node node) { // Ignore if node doesn't exist if (node == null) return; //找到适合的前继节点,当前节点的waitStatus赋值为CANCELLED
    node.thread = null; // 释放线程 // Skip cancelled predecessors 前驱节点已被取消  重新定义前驱节点
    Node pred = node.prev; //若前继节点是CANCELLED,则继续找前继节点,直至找到一个正常的前继节点赋值给node,作为node的新前继节点 while (pred.waitStatus > 0)
        node.prev = pred = pred.prev;

    Node predNext = pred.next;

    node.waitStatus = Node.CANCELLED; // 取消当前线程 所属的节点(标记为取消),  没有使用 cas  因为 其他线程 不会干扰这里 // If we are the tail, remove ourselves. //特殊情况:node==tail节点,将pred作为tail节点,然后将cancelledNodes节点链从CLH队列剔除 if (node == tail && compareAndSetTail(node, pred)) {
        compareAndSetNext(pred, predNext, null);
    } else { // If successor needs signal, try to set pred's next-link // so it will get one. Otherwise wake it up to propagate. int ws; //正常情况:则将cancelledNodes节点链从CLH队列剔除 if (pred != head &&
            ((ws = pred.waitStatus) == Node.SIGNAL ||
             (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
            pred.thread != null) {
            Node next = node.next; if (next != null && next.waitStatus <= 0)
                compareAndSetNext(pred, predNext, next);
        } else { //特殊情况:如果node是head的后继节点,则直接唤醒node的后继节点 pred==head节点:尝试调用unparkSuccessor(node),尝试唤醒当前节点的后继节点
            unparkSuccessor(node);
        }

        node.next = node; // help GC
    }
}

     6)unparkSuccessor(Node node);
     唤醒后继节点

/**
 * 唤醒后继节点
 * 注意:如果当前节点的后继节点为空,或者是被取消的节点。那就从tail节点逆向遍历CLH队列,直至找到一个距离当前节点node最近,且waitStatus<=0的节点,然后唤醒该节点
 */ private void unparkSuccessor(Node node) { /*
     * If status is negative (i.e., possibly needing signal) try
     * to clear in anticipation of signalling.  It is OK if this
     * fails or if status is changed by waiting thread.
     */ int ws = node.waitStatus; //获取头结点的等待状态 if (ws < 0) //把该状态设置成0
        compareAndSetWaitStatus(node, ws, 0); /*
     * 若后继节点不符合唤醒标准,则逆向遍历CLH,直至找到一个距离当前节点node最近,且waitStatus<=0的节点
     */
    Node s = node.next; //找到后继节点,唤醒后继节点 if (s == null || s.waitStatus > 0) { //很不巧,后继节点,节点为null,或者被取消
        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); //唤醒节点
}

3. release方法--释放资源

     1)boolean release(int arg);
     独占模式释放资源

/**
 * 资源的释放
 * 调用tryRelease方法进行释放锁
 * 释放锁成功后,获取头节点,接着唤醒后继节点,调用unparkSuccessor方法
 */ public final boolean release(int arg) { if (tryRelease(arg)) {
        Node h = head; if (h != null && h.waitStatus != 0)
            unparkSuccessor(h); return true;
    } return false;
}

     2)boolean tryRelease(int arg);
     独占模式下尝试释放锁,由子类选择性实现

protected boolean tryRelease(int arg) { throw new UnsupportedOperationException();
        }

     3)boolean releaseShared(int arg);
     共享模式释放资源

/**
 * Releases in shared mode.  Implemented by unblocking one or more
 * threads if {@link #tryReleaseShared} returns true.
 *
 * @param arg the release argument.  This value is conveyed to
 *        {@link #tryReleaseShared} but is otherwise uninterpreted
 *        and can represent anything you like.
 * @return the value returned from {@link #tryReleaseShared}
 */ public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) {
        doReleaseShared(); return true;
    } return false;
}

     4)boolean tryReleaseShared(int arg);
     共享模式下尝试释放锁,由子类选择性实现

protected boolean tryReleaseShared(int arg) { throw new UnsupportedOperationException();
        }

     5)int fullyRelease(Node node);
     使用当前节点状态调用release,成功返回状态,失败跑出异常

/**
 * 使用当前节点状态调用release,成功返回状态,失败跑出异常
 */ final int fullyRelease(Node node) { boolean failed = true; try { int savedState = getState(); if (release(savedState)) {
            failed = false; return savedState;
        } else { throw new IllegalMonitorStateException();
        }
    } finally { if (failed)
            node.waitStatus = Node.CANCELLED;
    }
}

4. CAS操作

/**
 * Unsafe类实例
 */ private static final Unsafe unsafe = Unsafe.getUnsafe(); /**  state内存偏移地址 */ private static final long stateOffset; /**  head内存偏移地址 */ private static final long headOffset; /** tail内存偏移地址 */ private static final long tailOffset; private static final long waitStatusOffset; /** next内存偏移地址 */ private static final long nextOffset; //静态初始化块 static { try { // 获取偏移量
        stateOffset = unsafe.objectFieldOffset
            (AbstractQueuedSynchronizer.class.getDeclaredField("state"));
        headOffset = unsafe.objectFieldOffset
            (AbstractQueuedSynchronizer.class.getDeclaredField("head"));
        tailOffset = unsafe.objectFieldOffset
            (AbstractQueuedSynchronizer.class.getDeclaredField("tail"));
        waitStatusOffset = unsafe.objectFieldOffset
            (Node.class.getDeclaredField("waitStatus"));
        nextOffset = unsafe.objectFieldOffset
            (Node.class.getDeclaredField("next"));

    } catch (Exception ex) { throw new Error(ex); }
} /**
 * CAS head field. Used only by enq.
 * 通过原子操作 改变上锁状态
 * this == null
 * 第一个参数为需要改变的对象,第二个为偏移量(即之前求出来的valueOffset的值),第三个参数为期待的值,第四个为更新后的值
 */ private final boolean compareAndSetHead(Node update) { //调用本地方法 实现硬件级别的原子操作 cas return unsafe.compareAndSwapObject(this, headOffset, null, update);
} /**
 * CAS tail field. Used only by enq.
 */ private final boolean compareAndSetTail(Node expect, Node update) { return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
} /**
 * CAS waitStatus field of a node.
 */ private static final boolean compareAndSetWaitStatus(Node node, int expect, int update) { return unsafe.compareAndSwapInt(node, waitStatusOffset,
                                    expect, update);
} /**
 * CAS next field of a node.
 */ private static final boolean compareAndSetNext(Node node,
                                               Node expect,
                                               Node update) { return unsafe.compareAndSwapObject(node, nextOffset, expect, update);
}

5. Condition条件队列

内部类ConditionObject,它实现了Condition接口,主要用于实现条件锁。
ConditionObject中也维护了一个队列,这个队列主要用于等待条件的成立,当条件成立时,其它线程将signal这个队列中的元素,将其移动到CLH的队列中,等待占有锁的线程释放锁后被唤醒。
Condition典型的运用场景是在BlockingQueue中的实现,当队列为空时,获取元素的线程阻塞在notEmpty条件上,一旦队列中添加了一个元素,将通知notEmpty条件,将其队列中的元素移动到AQS队列中等待被唤醒。

/**
 * 构造一个条件队列,来等待条件是否为真
 */ public class ConditionObject implements Condition, java.io.Serializable { /** 版本号 */ private static final long serialVersionUID = 1173984872572414699L; /** First node of condition queue. condition队列的头结点 */ private transient Node firstWaiter; /** Last node of condition queue. condition队列的尾结点 */ private transient Node lastWaiter; /**
     * Creates a new {@code ConditionObject} instance.
     * 构造函数
     */ public ConditionObject() { } /**
     * Adds a new waiter to wait queue.
     * @return its new wait node
     * 添加新的waiter到wait队列
     */ private Node addConditionWaiter() { //保存尾结点
        Node t = lastWaiter; // If lastWaiter is cancelled, clean out. 尾结点不为空,并且尾结点的状态不为CONDITION if (t != null && t.waitStatus != Node.CONDITION) { //清除状态为CONDITION的结点
            unlinkCancelledWaiters(); //将最后一个结点重新赋值给t
            t = lastWaiter;
        }
        Node node = new Node(Thread.currentThread(), Node.CONDITION); if (t == null)
            firstWaiter = node; else
            t.nextWaiter = node;
        lastWaiter = node; return node;
    }

AQS总结

1)AQS分为独占锁和共享锁。 2)AQS分为CLH自旋队列和Condition条件队列。 3)AQS是一个双向链表,由state状态控制。 4)AQS由volatile修饰保证多线程可见,采用CAS保证原子性。




如果您觉得本文的内容对您的学习有所帮助:支付鼓励



关键字:AQS源码分析
友荐云推荐