ReentrantLock源码揭秘

作者: 王侦 | 来源:发表于2022-11-11 21:53 被阅读0次

ReentrantLock是一种基于AQS框架的应用实现，是JDK中的一种线程并发访问的同步手段，它的功能类似于synchronized是一种互斥锁，可以保证线程安全。

相对于 synchronized， ReentrantLock具备如下特点：

可中断
可以设置超时时间
可以设置为公平锁
支持多个条件变量
与 synchronized 一样，都支持可重入

使用示例：

public class ReentrantLockDemo {

    private static int sum = 0;
    private static Lock lock = new ReentrantLock();

    public static void main(String[] args) throws InterruptedException {

        for (int i = 0; i < 3; i++) {
            Thread thread = new Thread(() -> {
                //加锁
                lock.lock();
                try {
                    for (int j = 0; j < 10000; j++) {
                        sum++;
                    }
                } finally {
                    // 解锁
                    lock.unlock();
                }
            });
            thread.start();
        }
        Thread.sleep(2000);
        System.out.println(sum);
    }
}

1.非公平锁加锁

默认情况下是非公平锁：

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

加锁源码如下：

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

来看看NonfairSync#lock：

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

加锁逻辑：

step1.这里首先会通过CAS尝试加锁，加锁成功则设置exclusiveOwnerThread为当前线程
step2.如果CAS加锁失败，则进入acquire(1)方法处理

acquire是在AQS里面定义的核心逻辑：

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

acquire加锁逻辑：

step1.tryAcquire(arg)尝试加锁，加锁成功则返回，加锁失败则继续后续逻辑
step2.addWaiter将加锁失败的线程构建成成Node节点，加入到同步等待队列
step3.acquireQueued这里会park线程，并且有线程被唤醒后尝试加锁的逻辑

注意上面addWaiter()和acquireQueued()逻辑AQS已经写好了，ReentrantLock主要实现了tryAcquire()，具体来看看这个方法

NonfairSync#tryAcquire()

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

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

从源码可以看出：

step1.通过CAS操作尝试加锁，加锁成功则设置exclusiveOwnerThread为当前线程
step2.实现了可重入锁加锁逻辑

2.非公平锁释放锁

这里非公平锁、公平锁都共用AQS里面的释放锁方法

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

调用AQS的释放锁方法：

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

释放锁逻辑：

step1.释放锁tryRelease(arg)
step2.释放锁成功，则唤醒同步队列head的后继节点的线程

此时在acquireQueued阻塞线程就会醒来，如果是头节点的后继节点，就会尝试加锁，加锁成功则返回，加锁失败则继续park阻塞。

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

另外，这里释放锁主要看ReentrantLock实现的tryRelease方法：

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

这里考虑到了可重入锁释放锁，只有当state减到0，才是真正释放锁成功，上面的release才能继续后面unpark唤醒阻塞的head后继节点的线程。

3.公平锁

加锁源码FairSync#lock：

       final void lock() {
            acquire(1);
        }

这里与非公平锁的区别就少了一个CAS尝试加锁的步骤，直接调用的就是acqire(1).

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

公平锁FairSync#tryAcquire：

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

从这里可以看出，只有当同步队列是空的时候，才会尝试加锁，否则就会直接入队（同步等待队列），根本不会尝试加锁，这就符合了公平锁的语义，遵守先来后到的顺序。

4.条件变量

调用Condition#await方法会释放当前持有的锁，然后阻塞当前线程，同时向Condition队列尾部添加一个节点，所以调用Condition#await方法的时候必须持有锁。

调用Condition#signal方法会将Condition队列的首节点移动到阻塞队列尾部，然后唤醒因调用Condition#await方法而阻塞的线程(唤醒之后这个线程就可以去竞争锁了)，所以调用Condition#signal方法的时候必须持有锁，持有锁的线程唤醒被因调用Condition#await方法而阻塞的线程。

使用示例：

/**
 * 条件变量
 */
@Slf4j
public class ReentrantLockDemo6 {
    private static ReentrantLock lock = new ReentrantLock();
    private static Condition cigCon = lock.newCondition();
    private static Condition takeCon = lock.newCondition();

    private static boolean hashcig = false;
    private static boolean hastakeout = false;

    //送烟
    public void cigratee() {
        lock.lock();
        try {
            while (!hashcig) {
                try {
                    log.debug("没有烟，歇一会");
                    cigCon.await();

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            log.debug("有烟了，干活");
        } finally {
            lock.unlock();
        }
    }

    //送外卖
    public void takeout() {
        lock.lock();
        try {
            while (!hastakeout) {
                try {
                    log.debug("没有饭，歇一会");
                    takeCon.await();

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            log.debug("有饭了，干活");
        } finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) {
        ReentrantLockDemo6 test = new ReentrantLockDemo6();
        new Thread(() -> {
            test.cigratee();
        }).start();

        new Thread(() -> {
            test.takeout();
        }).start();

        new Thread(() -> {
            lock.lock();
            try {
                hashcig = true;
                //唤醒送烟的等待线程
                cigCon.signal();
            } finally {
                lock.unlock();
            }


        }, "t1").start();

        new Thread(() -> {
            lock.lock();
            try {
                hastakeout = true;
                //唤醒送饭的等待线程
                takeCon.signal();
            } finally {
                lock.unlock();
            }
        }, "t2").start();
    }
}

4.1 await()

        public final void await() throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            Node node = addConditionWaiter();
            int savedState = fullyRelease(node);
            int interruptMode = 0;
            while (!isOnSyncQueue(node)) {
                LockSupport.park(this);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
            }
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null) // clean up if cancelled
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
        }

代码逻辑：

1）addConditionWaiter会将当前线程构建成节点Node加入到条件等待队列
2）fullyRelease完全释放锁
3）阻塞线程
4）后续就是线程被唤醒后者中断的逻辑

唤醒后的逻辑有两种情况

1）signal或者signalAll()唤醒，此时节点已经转移到同步队列，就会跳出while (!isOnSyncQueue(node)) 循环，进入acquireQueued(node, savedState) 尝试加锁和阻塞逻辑
2）其他线程中断了该线程checkInterruptWhileWaiting()，此时会将该节点加入到同步队列，但是没有从条件队列移除，所以后面有个补偿逻辑：node.nextWaiter != null时，unlinkCancelledWaiters()会从条件队列中删除

4.2 signalAll()

        public final void signalAll() {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            Node first = firstWaiter;
            if (first != null)
                doSignalAll(first);
        }

        private void doSignalAll(Node first) {
            lastWaiter = firstWaiter = null;
            do {
                Node next = first.nextWaiter;
                first.nextWaiter = null;
                transferForSignal(first);
                first = next;
            } while (first != null);
        }

    final boolean transferForSignal(Node node) {
        /*
         * If cannot change waitStatus, the node has been cancelled.
         */
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            return false;

        /*
         * Splice onto queue and try to set waitStatus of predecessor to
         * indicate that thread is (probably) waiting. If cancelled or
         * attempt to set waitStatus fails, wake up to resync (in which
         * case the waitStatus can be transiently and harmlessly wrong).
         */
        Node p = enq(node);
        int ws = p.waitStatus;
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
            LockSupport.unpark(node.thread);
        return true;
    }

核心逻辑：

1）将节点从条件队列移除
2）将节点加入到同步队列，并且唤醒线程，这里只有是取消节点或者cas设置节点状态为SIGNAL失败时，才会唤醒线程，其余情况不会

5.关于AQS

ReentrantLock就是基于AQS实现的。

AQS定义两种队列

同步等待队列：主要用于维护获取锁失败时入队的线程
条件等待队列：调用await()的时候会释放锁，然后线程会加入到条件队列，调用signal()唤醒的时候会把条件队列中的线程节点移动到同步队列中，等待再次获得锁

可见ReentrantLock符合经典的MESA模型：

ReentrantLock只用实现共享资源state的获取与释放方式即可，具体线程等待队列（同步等待队列、条件等待队列）的维护（如获取资源失败入队/唤醒出队等），AQS已经在顶层实现好了。

ReentrantLock主要实现如下方法：

tryAcquire(int)：独占方式。尝试获取资源，成功则返回true，失败则返回false。
tryRelease(int)：独占方式。尝试释放资源，成功则返回true，失败则返回false。

ReentrantLock源码揭秘

1.非公平锁加锁

2.非公平锁释放锁

3.公平锁

4.条件变量

4.1 await()

4.2 signalAll()

5.关于AQS

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读