并发编程之深入理解ReentrantLock和AQS原理

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

/** * Performs lock. Try immediate barge, backing up to normal * acquire on failure. */
final void lock() {
   
    //cas的操作保证原子性
    if (compareAndSetState(0, 1))
        //设置当前抢到锁的线程
        setExclusiveOwnerThread(Thread.currentThread());
    else
        acquire(1);
}

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

protected final boolean tryAcquire(int acquires) {
   
    return nonfairTryAcquire(acquires);
}

/** * Performs non-fair tryLock. tryAcquire is implemented in * subclasses, but both need nonfair try for trylock method. */
final boolean nonfairTryAcquire(int acquires) {
   
    final Thread current = Thread.currentThread();
    //获取state值
    int c = getState();
    //state=0表示当前没有线程持有锁，则使用cas尝试获取
    if (c == 0) {
   
        if (compareAndSetState(0, acquires)) {
   
            setExclusiveOwnerThread(current);
            //抢到了就退出
            return true;
        }
    }
    //如果state>0则表示当前锁被线程持有，则判断是不是自己持有
    else if (current == getExclusiveOwnerThread()) {
   
        //如果是当前线程，则重入，state+1
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

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;
}

protected final boolean tryRelease(int releases) {
   
    //直接将state-releases(允许一次释放多次，比如await方法，就会直接从state=n变成0)
    int c = getState() - releases;
    //当前释放锁的线程不是持有锁的线程则抛异常
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {
   
        free = true;
        //将持有锁的线程记录变量置为null
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}

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

/** * Creates and enqueues node for current thread and given mode. * * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared * @return the new node */
private Node addWaiter(Node mode) {
   
    //封装线程信息，并且mode为独占锁，ReentrantLock本来就是独占锁
    Node node = new Node(Thread.currentThread(), mode);
    // Try the fast path of enq; backup to full enq on failure
    Node pred = tail;
    if (pred != null) {
   
        node.prev = pred;
        //cas设置队尾
        if (compareAndSetTail(pred, node)) {
   
            pred.next = node;
            return node;
        }
    }
    //自旋cas加入队列
    enq(node);
    return node;
}

/** * Inserts node into queue, initializing if necessary. See picture above. * @param node the node to insert * @return node's predecessor */
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;
            if (compareAndSetTail(t, node)) {
   
                t.next = node;
                return t;
            }
        }
    }
}

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)) {
   
                //获取成功则将自己设置成头节点
                setHead(node);
                //将之前的头节点从队列中一处
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            //如果上面获取失败的话，这里就会判断是否需要阻塞，
            //主要是防止cpu无限调度这一块自旋代码，降低性能，从而使用通知的模式
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
   
        if (failed)
            cancelAcquire(node);
    }
}

/** * 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;
    if (ws == Node.SIGNAL)
        /* * This node has already set status asking a release * to signal it, so it can safely park. */
        return true;
    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. */
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

public final boolean release(int arg) {
   
    if (tryRelease(arg)) {
   
        Node h = head;
        if (h != null && h.waitStatus != 0)
            //唤醒阻塞的后继者
            unparkSuccessor(h);
        return true;
    }
    return false;
}

/** * Wakes up node's successor, if one exists. * * @param node the node */
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)
        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. */
    Node s = node.next;
    //如果存在后继节点，或者后继节点的状态为CANCELLED
    if (s == null || s.waitStatus > 0) {
   
        s = null;
        //从尾部开始取需要被唤醒的节点Node
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
    //存在需要唤醒的节点，则唤醒它
    if (s != null)
        LockSupport.unpark(s.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)) {
   
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            //判断是否需要阻塞
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
   
        if (failed)
            cancelAcquire(node);
    }
}