并发编程AQS--------ReentrantLock

Posted 2020-12-17 jihuifeng

同步框架AbstractQueuedSynchronizer

Java并发编程核心在于java.concurrent.util包 而juc当中的大多数同步器实现都是围绕着共同的基础行为，比如等待队列、条件队列、独占获取、共享获取等，而这个行为的抽象就是基于AbstractQueuedSynchronizer简称AQS，AQS定义了一套多线程访问共享资源的同步器框架，是一个依赖状态(state)的同步器。

AQS具备特性 阻塞等待队列、 共享/独占、 公平/非公平、 可重入(同一把锁同一个线程可重复拿)、 允许中断。

一般通过定义内部类Sync继承AQS 将同步器所有调用都映射到Sync对应的方法

 

 state 记录加锁的次数，体现锁的可重入性；访问方式哟有三种   getState()、setState()、compareAndSetState()

exclusiveOwnerThread 体现独占线程的特性，记录的是当前独占的线程。

AQS定义两种资源共享方式 Exclusive-独占，只有一个线程能执行，如ReentrantLock Share-共享，多个线程可以同时执行，如Semaphore/CountDownLatch

Node,阻塞队列的体现：等待队列，条件队列。各种属性定义如下

static final class Node {
    /**
     * 标记节点未共享模式
     * */
    static final Node SHARED = new Node();
    /**
     *  标记节点为独占模式
     */
    static final Node EXCLUSIVE = null;

    /**
     * 在同步队列中等待的线程等待超时或者被中断，需要从同步队列中取消等待
     * */
    static final int CANCELLED =  1;
    /**
     *  后继节点的线程处于等待状态，而当前的节点如果释放了同步状态或者被取消，
     *  将会通知后继节点，使后继节点的线程得以运行。
     */
    static final int SIGNAL    = -1;
    /**
     *  节点在等待队列中，节点的线程等待在Condition上，当其他线程对Condition调用了signal()方法后，
     *  该节点会从等待队列中转移到同步队列中，加入到同步状态的获取中
     */
    static final int CONDITION = -2;
    /**
     * 表示下一次共享式同步状态获取将会被无条件地传播下去
     */
    static final int PROPAGATE = -3;

    /**
     * 标记当前节点的信号量状态 (1,0,-1,-2,-3)5种状态
     * 使用CAS更改状态，volatile保证线程可见性，高并发场景下，
     * 即被一个线程修改后，状态会立马让其他线程可见。
     */
    volatile int waitStatus;

    /**
     * 前驱节点，当前节点加入到同步队列中被设置
     */
    volatile Node prev;

    /**
     * 后继节点
     */
    volatile Node next;

    /**
     * 节点同步状态的线程
     */
    volatile Thread thread;

    /**
     * 等待队列中的后继节点，如果当前节点是共享的，那么这个字段是一个SHARED常量，
     * 也就是说节点类型(独占和共享)和等待队列中的后继节点共用同一个字段。
     */
    Node nextWaiter;

如果节点存在条件队列中，只能是独占模式不能是共享方式。

等待队列其实是一个双向链表结构，每个节点记录的有前驱节点和后驱节点。

两种队列都是基于Node节点构建的，每个节点的Node都会有信号量，也就是属性waitSate

 

 条件队列其实是单向链表；

 ReentrantLock是一把可重入的，独占的显示锁，构造的时候可以传入一个布尔值来决定是不是公平/非公平锁

public ReentrantLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
    }

true为公平锁，false为非公平锁

如何理解公平锁和非公平锁

假如线程A结束后唤醒了线程B，此时新来一个线程C，如果线程C能和线程B抢锁，那么这个是非公平锁，如果新来的线程C只能怪怪去排队，那么就是公平锁。

加锁过程，每加一次锁state都会加1，释放一次锁，state都会减1，exclusiveOwnerThread记录的当前持有锁的线程。

 

 

 分析一下取锁的源码

 protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {--------表示没有线程占用锁，可以取拿锁
                if (!hasQueuedPredecessors() &&  -------------判断队列里面没有线程在等待才能去抢锁，这就是公平的体现
                    compareAndSetState(0, acquires)) {  -------cas算法原子操作改变state值，state值又被volitale修饰，保证并发下修改state的安全性。
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            else if (current == getExclusiveOwnerThread()) {
                int nextc = c + acquires;
                if (nextc < 0)
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }
    }

 

 线程被阻塞之后会被放在等待队列里面，即我们那个双链表结够，注意我们这个双链表的head的属性thread是null，也就是说head不放任何线程信息，仅仅是做head标记位使用。

 

 

 ReentrantLock加锁的过程

 public final void acquire(int arg) {
        // 尝试获取锁，获取锁失败，addWaiter方法加入到CLH等待队列，Node.EXCLUSIVE表示以独占的方式入队；
        if (!tryAcquire(arg) &&
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

private Node addWaiter(Node mode) {
    // 1. 将当前线程构建成Node类型
    Node node = new Node(Thread.currentThread(), mode);
    // Try the fast path of enq; backup to full enq on failure
    Node pred = tail;
    // 2. 1当前尾节点是否为null？
    if (pred != null) {
        // 2.2 将当前节点尾插入的方式
        node.prev = pred;
        // 2.3 CAS将节点插入同步队列的尾部
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    enq(node);
    return node;
}


注意这里的for(;;)循环；

private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        if (t == null) { // Must initialize
            //队列为空需要初始化，创建空的头节点
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            node.prev = t;
            //set尾部节点
            if (compareAndSetTail(t, node)) {//当前节点置为尾部
                t.next = node; //前驱节点的next指针指向当前节点
                return t;
            }
        }
    }
}

注意AQS里面的线程唤醒不会唤醒所有的等待线程，而回唤醒头节点的next线程（head头节点不放线程），做到顺序唤醒，而object的notify方法和notifyall方法会唤醒所以有的线程。无序。

线程的阻塞和唤醒用的是魔术类Unsafe里面的park（）和unpark（）方法，底层是调用Pthead_mutex_lock指令库方法。

代码;

 public final void acquire(int arg) {
        // 尝试获取锁，获取锁失败，addWaiter方法加入到CLH等待队列，Node.EXCLUSIVE表示以独占的方式入队；
        // acquireQueued对入队的线程进行阻塞
        if (!tryAcquire(arg) &&
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

/**
 * 已经在队列当中的Thread节点，准备阻塞等待获取锁
 */
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)) {//如果前驱结点是头结点，才tryAcquire，其他结点是没有机会tryAcquire的。
                setHead(node);//获取同步状态成功，将当前结点设置为头结点。
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            /**
             * 如果前驱节点不是Head，通过shouldParkAfterFailedAcquire判断是否应该阻塞
             * 前驱节点信号量为-1，当前线程可以安全被parkAndCheckInterrupt用来阻塞线程
             */
            if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

public class LockSupport {
    private LockSupport() {} // Cannot be instantiated.

    private static void setBlocker(Thread t, Object arg) {  //unsafe魔术类阻塞线程。
        // Even though volatile, hotspot doesn‘t need a write barrier here.
        UNSAFE.putObject(t, parkBlockerOffset, arg);
    }