ReentrantLock源码解析

2019-03-31  本文已影响0人  binecy

要理解ReentrantLock,首先要理解AbstractQueuedSynchronizer。
AbstractQueuedSynchronizer类如其名,抽象的队列式的同步器,AQS定义了一套多线程访问共享资源的同步器框架,许多同步类实现都依赖于它,如常用的ReentrantLock/Semaphore/CountDownLatch。

image

它维护了一个volatile int state(代表共享资源)和一个FIFO线程等待队列(多线程争用资源被阻塞时会进入此队列)。
acquire方法表示申请资源,它调用tryAcquire方法(tryAcquire由子类实现),如果tryAcquire失败,会将当前线程转化为等待队列的一个节点。
release方法表示释放资源,它调用tryRelease(tryRelease由子类实现),如果tryRelease成功,会唤醒等待队列某个节点的线程。
除了tryAcquire/tryRelease,AbstractQueuedSynchronizer还提供了tryAcquireShared/tryReleaseShared/ isHeldExclusively。

不同的同步类正是通过这些方法实现不同逻辑。

ReentrantLock实现了公平锁FairSync和非公平锁NonfairSync,它们继续自ReentrantLock内部抽象类Sync,Sync实现了一些公用的逻辑。

这篇文章只关注NonfairSync。

单线程加锁

查看NonfairSync.lock

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

compareAndSetState是AbstractQueuedSynchronizer提供的方法,通过CAS争夺AbstractQueuedSynchronizer的资源state。

    protected final boolean compareAndSetState(int expect, int update) {
        // See below for intrinsics setup to support this
        return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
    }

单线程环境下,compareAndSetState(0, 1)可以成功,setExclusiveOwnerThread设置持有锁的线程为当前线程。
否则调用AbstractQueuedSynchronizer.acquire重新争夺资源。

单线程解锁

unlock调用AbstractQueuedSynchronizer.release,它会先调用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);  // 持有锁的线程设置为null
            }
            setState(c);  // 更新State
            return free;
        }

单线程环境下,只需要将state减去releases的值,State为0时,清除当前持锁线程就行。

setState没有使用CAS操作,因为当前线程已经占有资源(state),注意:state是volatile的。

多线程环境,可能还要唤醒等待队列中的线程,后面再说。

多线程抢锁

抢锁的关键在NonfairSync.lock方法中的compareAndSetState(0, 1)操作,CAS保证只有一个线程抢锁成功,其他线程将调用AbstractQueuedSynchronizer.acquire

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

tryAcquire是AbstractQueuedSynchronizer提供的抽象方法,实现在Sync.nonfairTryAcquire

        final boolean nonfairTryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {  // state == 0,重新抢锁
                if (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;
        }

如果state不是0而且持有锁的非当前线程,返回false。AbstractQueuedSynchronizer就将当前线程转化为一个Waiter,添加到等待队列中。

回到addWaiter(Node.EXCLUSIVE), arg),EXCLUSIVE表示独占模式

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

enq方法将node插入等待队列,如果head/tail为null会初始化他们。
如果compareAndSetTail(pred, node)的CAS操作失败(多个线程操作),也会进入enq方法。

private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        if (t == null) { // 初始化tail/head
            if (compareAndSetHead(new Node()))  
                tail = head;
        } else {
            node.prev = t;
            if (compareAndSetTail(t, node)) { // 更新tail
                t.next = node;
                return t;
            }
        }
    }
}

可以看到,head是一个标识节点,它不代表一个请求锁的线程。
这里for (;;)可以保证初始化head/tail或更新tail操作成功,否则不断重试。

addWaiter完成,回来看看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)) {  // 前一个节点是head,重新抢锁
                setHead(node);  // 抢锁成功,设置head为node
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

如果某个节点抢锁成功,会将这个节点设置为head。

如果抢锁失败,调用shouldParkAfterFailedAcquire

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            return true;
        if (ws > 0) {  // ws = CANCELLED
            // CODE1
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            // CODE2
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

waitStatus是一个等待队列节点的内部状态,有
SIGNAL:该节点的后继节点(或将)被阻塞,因此当前节点释放或取消时必须唤醒其后继节点。
CANCELLED:由于超时或中断,此节点被取消。
CONDITION/PROPAGATE暂时不关注。
0是默认状态。

只有CANCELLED状态大于0,pred.waitStatus > 0表示pred为已取消节点。所以CODE1处理会清除当前节点前的CANCELLED状态节点。
CODE2会更改prev节点的状态为SIGNAL。
如果pred.waitStatus不是SIGNAL,shouldParkAfterFailedAcquire会在acquireQueued的for循环中重复调用,直到返回true。

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

LockSupport.park(this);会挂起当前线程,阻塞等待。
这里返回并清除线程的中断信息,最后acquireQueued方法将线程中断信息interrupted返回。
回到acquire方法,如果acquireQueued返回true,将中断当前线程。

多线程解锁

ReentrantLock.unlock调用AbstractQueuedSynchronizer.方法

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

head状态不能为CANCELLED,
head不为null而且waitStatus不是初始状态0时,调用unparkSuccessor,唤醒head的下一个节点线程

private void unparkSuccessor(Node node) {
    int ws = node.waitStatus;
    if (ws < 0)  // 清除当前节点waitStatus
        compareAndSetWaitStatus(node, ws, 0);

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

如果s为null或s.waitStatus为CANCELLED,则从tail向前遍历,找到最前面一个waitStatus不为CANCELLED的节点。
LockSupport.unpark(s.thread)唤醒LockSupport.park挂起的线程。
这时前面阻塞的线程回到acquireQueued方法,重新执行for (;;)循环,重新tryAcquire。
如果其他线程此时也调用lock方法,可能其他线程加锁成功,被唤醒的线程重新阻塞(抢锁的关键在于compareAndSetState(0, 1))。

tryLock

tryLock调用到AbstractQueuedSynchronizer.tryAcquireNanos

    public final boolean tryAcquireNanos(int arg, long nanosTimeout)
            throws InterruptedException {
        if (Thread.interrupted())  //  获取并清除中断信息
            throw new InterruptedException();
        return tryAcquire(arg) ||
            doAcquireNanos(arg, nanosTimeout);
    }

同样,tryAcquire失败后,调用doAcquireNanos:

    private boolean doAcquireNanos(int arg, long nanosTimeout)
            throws InterruptedException {
        if (nanosTimeout <= 0L)
            return false;
        final long deadline = System.nanoTime() + nanosTimeout;  // 超时时间
        final Node node = addWaiter(Node.EXCLUSIVE);
        boolean failed = true;
        try {
            for (;;) {
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {  // 重新抢锁
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return true;
                }
                nanosTimeout = deadline - System.nanoTime();
                if (nanosTimeout <= 0L)  // 已超时
                    return false;
                if (shouldParkAfterFailedAcquire(p, node) &&
                    nanosTimeout > spinForTimeoutThreshold)
                    LockSupport.parkNanos(this, nanosTimeout);  // 超时阻塞
                if (Thread.interrupted())  // 已中断
                    throw new InterruptedException();
            }
        } finally {
            // 超时,撤销操作
            if (failed)
                cancelAcquire(node);
        }
    }

和acquireQueued方法类似。
nanosTimeout > spinForTimeoutThreshold时,不会挂起线程,而是不断tryAcquire,这里是自旋锁。
spinForTimeoutThreshold是1000纳秒。

cancelAcquire

CANCELLED状态在cancelAcquire方法中产生,如果加锁异常,中断或超时会调用这个方法

    private void cancelAcquire(Node node) {
        if (node == null)
            return;

        node.thread = null;

        // 跳过前面CANCELLED的节点
        Node pred = node.prev;
        while (pred.waitStatus > 0)
            node.prev = pred = pred.prev;

        Node predNext = pred.next;
        // 修改waitStatus为CANCELLED
        node.waitStatus = Node.CANCELLED;

        // 如果是tail,移除该节点
        if (node == tail && compareAndSetTail(node, pred)) {
            compareAndSetNext(pred, predNext, null);
        } else {
            int ws;
            if (pred != head &&
                ((ws = pred.waitStatus) == Node.SIGNAL ||
                 (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
                pred.thread != null) {
                Node next = node.next;
                if (next != null && next.waitStatus <= 0)
                    // 修改next-link
                    compareAndSetNext(pred, predNext, next);
            } else {
                unparkSuccessor(node);  // 唤醒下一个节点
            }

            node.next = node; // help GC
        }
    }

如果pred节点是head(本来该当前节点运行了),直接唤醒当前节点的下一个节点。
否则通过CAS操作更新pred节点的WaitStatus为SIGNAL,成功则修改pred节点的next属性指向下一个节点,失败(可能pred节点已经变化了),唤醒当前节点的下一个节点。

可中断锁

lockInterruptibly方法调用AbstractQueuedSynchronizer.doAcquireInterruptibly,这个方法和acquireQueued差不多

    private void doAcquireInterruptibly(int arg)
        throws InterruptedException {
        final Node node = addWaiter(Node.EXCLUSIVE);
        boolean failed = true;
        try {
            for (;;) {
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();  // 抛出异常,中断锁
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

注意:LockSupport.park是可以响应中断的。

上一篇 下一篇

猜你喜欢

热点阅读