乐趣区

关于java:AQSReentrantLock源码

前言

AbstractQueuedSynchronizer(形象同步队列器),因为它是 Java 并发包的根底工具类,是实现 ReentrantLock、CountDownLatch、Semaphore、FutureTask 等类的根底。## 构造

// 头结点,你间接把它当做 以后持有锁的线程 可能是最好了解的
private transient volatile Node head;

// 阻塞的尾节点,每个新的节点进来,都插入到最初,也就造成了一个链表
private transient volatile Node tail;

// 这个是最重要的,代表以后锁的状态,0 代表没有被占用,大于 0 代表有线程持有以后锁
// 这个值能够大于 1,是因为锁能够重入,每次重入都加上 1
private volatile int state;

// 代表以后持有独占锁的线程,举个最重要的应用例子,因为锁能够重入
// reentrantLock.lock()能够嵌套调用屡次,所以每次用这个来判断以后线程是否曾经领有了锁
// if (currentThread == getExclusiveOwnerThread()) {state++}
private transient Thread exclusiveOwnerThread; // 继承自 AbstractOwnableSynchronizer

留神:阻塞队列不蕴含 head

期待队列中的每个线程都被包装为一个 Node 实例,构造是链表

 static final class Node {
        /** Marker to indicate a node is waiting in shared mode */
         // 标识以后节点在共享模式下
        static final Node SHARED = new Node();
        /** Marker to indicate a node is waiting in exclusive mode */
         // 标识以后节点在独占模式下
        static final Node EXCLUSIVE = null;

        /** waitStatus value to indicate thread has cancelled. */
         // 代表此线程勾销了争抢这个锁
        static final int CANCELLED =  1;
        /** waitStatus value to indicate successor's thread needs unparking. */
         // 代表以后 node 的后继节点对应的线程须要被唤醒
        static final int SIGNAL    = -1;
        /** waitStatus value to indicate thread is waiting on condition. */
         // 代表以后线程在期待 condition
        static final int CONDITION = -2;
        /**
         * waitStatus value to indicate the next acquireShared should
         * unconditionally propagate.
         */
        static final int PROPAGATE = -3;

       
         // 如果这个值 大于 0 代表此线程勾销了期待,volatile int waitStatus;

         // 前驱节点
        volatile Node prev;

         // 后继节点
        volatile Node next;

           // 线程
        volatile Thread thread;

           // 下个期待节点
        Node nextWaiter;

    }

ReentrantLock 源码剖析

构造函数

// 非偏心锁
public ReentrantLock() {sync = new NonfairSync();
}

public ReentrantLock(boolean fair) {sync = fair ? new FairSync() : new NonfairSync();}
//ReentrantLock 在外部用了外部类 Sync 来治理锁,所以真正的获取锁和开释锁是由 Sync 的实现类来管制的。abstract static class Sync extends AbstractQueuedSynchronizer {}

线程抢锁(偏心锁)

    //lock 办法
    public void lock() {sync.acquire(1);
    } 
    
    
    public final void acquire(int arg) {
        // 此时 arg=1
        // 如果过 tryAcquire 办法返回为 true,则此办法间接完结,否则调用 acquireQueued 将线程压入到队列之中
        if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();}


static final class FairSync extends Sync {
        private static final long serialVersionUID = -3000897897090466540L;
        /**
         * Fair version of tryAcquire.  Don't grant access unless
         * recursive call or no waiters or is first.
         */
        // 返回 boolean 值,代表是否获取到锁
        // 有两种状况返回 true(1. 没有线程在期待锁。2. 重入锁,线程自身就持有锁)@ReservedStackAccess
        protected final boolean tryAcquire(int acquires) {final Thread current = Thread.currentThread();
            int c = getState();
            //state 为 0,阐明此时此刻,没有线程持有锁
            if (c == 0) {
                // 尽管此时此刻锁是能够用的,然而这是偏心锁。// 须要判断队列中是否有其余线程曾经在期待
                if (!hasQueuedPredecessors() &&
                    // 如果没有线程在期待,就 CAS 获取锁(即通过 CAS 扭转)// 不胜利的话,阐明就在刚刚简直同一时刻有个线程领先了 
                    compareAndSetState(0, acquires)) {
                       // 抢占胜利,对锁进行标识。setExclusiveOwnerThread(current);
                    return true;
                }
            }
            // 阐明锁重入了,须要对 state 进行 +1 操作
            else if (current == getExclusiveOwnerThread()) {
                int nextc = c + acquires;
                if (nextc < 0)
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            // 阐明没有获取到锁,将会执行 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) 
            return false;
        }
    
    
     /**
     * Creates and enqueues node for current thread and given mode.
     *
     * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared
     * @return the new node
     */
    // 在执行 acquireQueued 之前,先会调用 addWaiter(Node.EXCLUSIVE), arg) 办法
    // 此办法会将以后线程包装为 Node,同时进入到队列中
    // 参数 mode 此时是 Node.EXCLUSIVE,代表独占模式
    private Node addWaiter(Node mode) {// 新建 Node,在构造方法中通过 Thread.currentThread()将以后线程封装。具体细节可看 Node 的构造方法
        Node node = new Node(mode);
        
        // 采纳自旋形式入队
        for (;;) {
            // 获取以后 AQS 的尾节点
            Node oldTail = tail;
            // 如果以后尾节点不为 null
            if (oldTail != null) {
                // 将队列尾节点设置为新节点的 prev
                node.setPrevRelaxed(oldTail);
                // 将以后线程排到队尾,有线程竞争就反复排
                if (compareAndSetTail(oldTail, node)) {
                    oldTail.next = node;
                       // 将以后 node 返回
                    return node;
                }
            } else {
                // 初始化队列,初始化实现后,进入下面的 if 代码中
                initializeSyncQueue();}
        }
    }
    
    // 可能存在线程竞争,应用 CAS 进行初始化
    // 初始化队列时,可看出并没有设置队列的 head。private final void initializeSyncQueue() {
        Node h;
        if (HEAD.compareAndSet(this, null, (h = new Node())))
            tail = h;
    }
    
    
     /**
     * Acquires in exclusive uninterruptible mode for thread already in
     * queue. Used by condition wait methods as well as acquire.
     * 以独占模式获取在队列中的线程
     * @param node the node 曾经进入阻塞队列的 node
     * @param arg the acquire argument
     * @return {@code true} if interrupted while waiting 
     */
    // if (!tryAcquire(arg) 
    //        && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) 
    //     selfInterrupt();
    // 如果此办法返回为 true,意味着下面的代码将会执行 selfInterrupt() 办法
    // 真正的线程挂起,以及唤醒线程获取锁
    final boolean acquireQueued(final Node node, int arg) {
        boolean interrupted = false;
        try {for (;;) {// p 是以后节点的前驱节点,并且 p 节点不可能为 null。具体见 predecessor()办法
                final Node p = node.predecessor();
                // 如果以后节点是阻塞队列的第一个节点,会去尝试获取锁
                // 阻塞队列不蕴含 head 节点,head 个别指的是占有锁的线程,head 前面的才称为阻塞队列。if (p == head && tryAcquire(arg)) {setHead(node);
                    p.next = null; // help GC
                    return interrupted;
                }
                // 以后节点不是对头,或者抢占失败
                // 看办法名可知,判断是否须要将以后线程挂起
                if (shouldParkAfterFailedAcquire(p, node))
                    // 如果 parkAndCheckInterrupt()为 true,则将 interrupted 变为 true
                    interrupted |= parkAndCheckInterrupt();}
        } catch (Throwable t) {cancelAcquire(node);
            if (interrupted)
                selfInterrupt();
            throw t;
        }
    }
    
     /**
     * Checks and updates status for a node that failed to acquire.
     * Returns true if thread should block. This is the main signal
     * control in all acquire loops.  Requires that pred == node.prev.
     *
     * @param pred node's predecessor holding status
     * @param node the node
     * @return {@code true} if thread should block
     */
    // 以后线程没有抢到锁,是否须要挂起以后线程?private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        // 前驱节点 waitStatus 为 1,阐明前驱节点状态失常,以后线程须要挂起。返回 true 即可
        if (ws == Node.SIGNAL)
            /*
             * This node has already set status asking a release
             * to signal it, so it can safely park.
             */
            return true;
        // 前驱节点 waitStatus 大于 0,之前说过,大于 0 阐明前驱节点勾销了排队。// 这里须要晓得这点:进入阻塞队列排队的线程会被挂起,而唤醒的操作是由前驱节点实现的。// 所以上面这块代码说的是将以后节点的 prev 指向 waitStatus<= 0 的节点,// 简略说,就是为了找个好爹,因为你还得依赖它来唤醒呢,如果前驱节点勾销了排队,// 找前驱节点的前驱节点做爹,往前遍历总能找到一个好爹的
        if (ws > 0) {
            /*
             * Predecessor was cancelled. Skip over predecessors and
             * indicate retry.
             */
            do {node.prev = pred = pred.prev;} while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            /*
             * waitStatus must be 0 or PROPAGATE.  Indicate that we
             * need a signal, but don't park yet.  Caller will need to
             * retry to make sure it cannot acquire before parking.
             */
            // 进入此处意味着,前驱节点的 waitStatus 不等于 - 1 和 1,那也就是只可能是 0,-2,-3
            // 每个新入队的 Node 的 waitStatus 都是 0,失常状况下,前驱节点是之前的 tail,那么它的 waitStatus 应该是 0
            // 用 CAS 将前驱节点的 waitStatus 设置为 Node.SIGNAL(也就是 -1)
            pred.compareAndSetWaitStatus(ws, Node.SIGNAL);
        }
        // 返回 false 之后,会在 acquireQueued()办法的 for 循环中再次进入此办法,不过会进入第一个分支
        return false;
    }
    // 在第一次进入 shouldParkAfterFailedAcquire()办法时,通常会返回 false
    // 起因很简略,前驱节点的 waitStatus=- 1 是依赖于后继节点设置的。// 并且这个办法是套在循环里的,所以第二次进来的时候状态就是 - 1 了
    
     /**
     * Convenience method to park and then check if interrupted.
     *
     * @return {@code true} if interrupted
     */
    // 挂起线程,期待被唤醒
    private final boolean parkAndCheckInterrupt() {LockSupport.park(this);
        // Thread.interrupted() 默认返回 false
        return Thread.interrupted();}
    
 
}

加锁(偏心锁)留神点

  • 阻塞队列不蕴含 head 头节点,head 头节点是持有以后锁的线程
  • 线程被挂起时,是须要前驱节点对其进行唤醒

线程解锁(偏心锁)

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

public final boolean release(int arg) {if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

// 回到 ReentrantLock 看 tryRelease 办法
protected final boolean tryRelease(int releases) {int c = getState() - releases;
    // 判断以后线程是否持有锁
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    // 是否齐全开释锁
    boolean free = false;
    // 其实就是重入的问题,如果 c ==0,也就是说没有嵌套锁了,能够开释了,否则还不能开释掉
    if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}

/**
 * Wakes up node's successor, if one exists.
 *
 * @param node the node
 */
// 唤醒后继节点
// 从下面调用处晓得,参数 node 是 head 头结点
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;
    // 如果 head 节点以后 waitStatus<0, 将其批改为 0
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);
    /*
     * Thread to unpark is held in successor, which is normally
     * just the next node.  But if cancelled or apparently null,
     * traverse backwards from tail to find the actual
     * non-cancelled successor.
     */
    // 上面的代码就是唤醒后继节点,然而有可能后继节点勾销了期待(waitStatus==1)// 从队尾往前找,找到 waitStatus<= 0 的所有节点中排在最后面的
    Node s = node.next;
    if (s == null || s.waitStatus > 0) {
        s = null;
        // 从后往前找,认真看代码,不用放心两头有节点勾销 (waitStatus==1) 的状况
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
    if (s != null)
        // 唤醒线程
        LockSupport.unpark(s.thread);
}


private final boolean parkAndCheckInterrupt() {LockSupport.park(this); // 刚刚线程被挂起在这里了
    return Thread.interrupted();}
// 又回到这个办法了:acquireQueued(final Node node, int arg),这个时候,node 的前驱是 head 了,持续去尝试获取锁

初步总结

在并发环境下,加锁和解锁须要以下三个部件的协调:

  1. 锁状态。state 的作用就是判断锁是否被别的线程占有了,它为 0 的时候代表没有线程占有锁,能够去争抢这个锁,用 CAS 将 state 设为 1,如果 CAS 胜利,阐明抢到了锁,这样其余线程就抢不到了,如果锁重入的话,state 进行 +1 就能够,解锁就是减 1,直到 state 又变为 0,代表开释锁,所以 lock() 和 unlock() 必须要配对啊。而后唤醒期待队列中的第一个线程,让其来占有锁。
  2. 线程的阻塞和解除阻塞。AQS 中采纳了 LockSupport.park(thread) 来挂起线程,用 unpark 来唤醒线程。
  3. 阻塞队列。因为争抢锁的线程可能很多,然而只能有一个线程拿到锁,其余的线程都必须期待,这个时候就须要一个 queue 来治理这些线程,AQS 用的是一个 FIFO 的队列,就是一个链表,每个 node 都持有后继节点的援用。
退出移动版