【JUC系列】ReentrantLock实现本地锁的源码分析

使用场景

public class ReentrantLockTest {
    private static ReentrantLock lock = new ReentrantLock();

    public static void main(String[] args) {
        new Thread(()->{
            lock.lock();
            // do something
            System.out.println("111");
            try {
                Thread.sleep(Integer.MAX_VALUE);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            lock.unlock();
        }).start();

        new Thread(()->{
            lock.lock();
            // do something
            try {
                Thread.sleep(Integer.MAX_VALUE);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("222");
            lock.unlock();
        }).start();
    }
}

源码分析

默认使用非公平锁。

    public ReentrantLock() {
        sync = new NonfairSync();
    }

加锁

ReentrantLock#lock加锁使用sync进行加锁。

public void lock() {
    sync.lock();
}

如果当前的状态为0则设置exclusiveOwnerThread = thread;表示当前线程占住当前锁对象。

final void lock() {
    if (compareAndSetState(0, 1))
        setExclusiveOwnerThread(Thread.currentThread());
    else
        acquire(1);
}

AbstractQueuedSynchronizer#acquire。

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

非公平锁的实现ReentrantLock.NonfairSync#tryAcquire。

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

ReentrantLock.Sync#nonfairTryAcquire。当前状态值为0直接获取锁如果锁是当前线程占用则是可重复锁state++。

        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()) {
                int nextc = c + acquires;
                if (nextc < 0) // overflow
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }

如果没有获取到锁在队列中等待执行AbstractQueuedSynchronizer#addWaiter。末尾节点不为空设置末尾节点为node的前置节点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
        Node pred = tail;
        if (pred != null) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        enq(node);
        return node;
    }

末尾节点为空执行AbstractQueuedSynchronizer#enq。进行末尾节点的初始化再插入node节点。

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

AbstractQueuedSynchronizer#acquireQueued。如果上一个节点是头结点尝试获取锁。

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

AbstractQueuedSynchronizer#shouldParkAfterFailedAcquire前置节点可唤醒返回true否则设置前置节点的等待状态为-1。

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

AbstractQueuedSynchronizer#parkAndCheckInterrupt找到唤醒自己的进入休眠。

private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
}

解锁

ReentrantLock#unlock。

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

AbstractQueuedSynchronizer#release。如果成功释放锁且头节点不为空waitStatus不为0执行AbstractQueuedSynchronizer#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;
}

ReentrantLock.Sync#tryRelease修改state-1如果state=0则释放exclusiveOwnerThread返回true否则返回false。

        protected final boolean tryRelease(int releases) {
            int c = getState() - releases;
            if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
            boolean free = false;
            if (c == 0) {
                free = true;
                setExclusiveOwnerThread(null);
            }
            setState(c);
            return free;
        }

AbstractQueuedSynchronizer#unparkSuccessor。从尾结点找到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)
        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;
    if (s == null || s.waitStatus > 0) {
        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);
}

公平锁和非公平锁的不同在于tryAcquire方法。核心在于AbstractQueuedSynchronizer#hasQueuedPredecessors该方法判断锁同步队列没有多余的节点。

protected final boolean tryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        if (!hasQueuedPredecessors() &&
            compareAndSetState(0, acquires)) {
            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;
}

总结概括

1. 加锁的逻辑
   • 没有锁state由0变成1设置exclusiveOwnerThread = thread;
   • 有锁且拥有锁的线程和加锁线程是同一个state++
   • 有锁但拥有锁的线程和加锁线程不是用一个队尾添加节点设置前置节点的waitStatus=-1进行休眠。
2. 解锁的逻辑。state–当state=0释放锁成功设置exclusiveOwnerThread = null;唤醒最后一个waitStatus<=0。