源码分析----ReentrantLock实现和AbstractQueuedSynchronizer

时间：2022-07-04 18:37:15

ReentrantLock有公平锁和非公平锁，默认是非公平锁，而其加锁的解锁的操作其实都是依赖于某个内部对象

public class ReentrantLock implements Lock, java.io.Serializable {
private final Sync sync;

abstract static class Sync extends AbstractQueuedSynchronizer {
}

static final class NonfairSync extends Sync {
}

static final class FairSync extends Sync {
}

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

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

}
Sync继承于AbstractQueuedSynchronizer，锁的操作都是依赖于AbstractQueuedSynchronizer

AbstractQueuedSynchronizer中有一个state的变量，0代表没有锁没有被获取，1代表获取了锁，AQS提供了一个线程安全的方法compareAndSetState让我们设置state的值


然后分析一下lock方法，lock方法调用的是NonfairSync的lock方法

    final void lock() {
        if (compareAndSetState(0, 1))//设置state的值为1，设置成功则代表获取到锁并返回true
            setExclusiveOwnerThread(Thread.currentThread());//设置当前线程拥有锁
        else
            acquire(1);
    }
acquire是AQS中的方法 
    public final void acquire(int arg) {
        if (!tryAcquire(arg) &&//尝试获取锁，获取成功设返回true
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

    protected boolean tryAcquire(int arg) {
        throw new UnsupportedOperationException();
    }
可以看到tryAcquire抛出了异常，这个方法主要是子类覆盖使用，子类来判断是非可以获取锁，所以这个放在在 NonfairSync中实现，在这里NonfairSync的tryAcquire调用的是Sync的nonfairTryAcquire方法 
final boolean nonfairTryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {//如果没有线程获取锁
            if (compareAndSetState(0, acquires)) {//尝试获取锁，使用AQS提供的方法设置state
                setExclusiveOwnerThread(current);//成功则设置占有锁的是当前线程
                return true;
            }
        }
        //由于ReentrantLock是可重入锁，所以需要判断当前占有锁的是否是当前线程
        else if (current == getExclusiveOwnerThread()) {
            int nextc = c + acquires;//state加1
            if (nextc < 0) 
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false;
    }
如果失败了就会创建一个节点(代表当前线程)放入队列并阻塞

创建节点是addWaiter方法

    private Node addWaiter(Node mode) {
    //ReentrantLock是排他锁，所以结点类型是Node.EXCLUSIVE
        Node node = new Node(Thread.currentThread(), mode);
        Node pred = tail;
        //接下来就是将结点加入到队列尾部
        if (pred != null) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {//使用线程安全的方法设置
                pred.next = node;
                return node;
            }
        }
        //当compareAndSetTail失败的时候会调用enq方法
        enq(node);
        return node;
    }

    private Node enq(final Node node) {
        for (;;) {//循环中不断重试
            Node t = tail;
            if (t == null) { //如果为空则为第一个结点
                if (compareAndSetHead(new Node()))//创建一个节点作为头结点
                    tail = head;
            } else {//接下来就是将结点加到尾部直至成功
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }
当节点成功加入后，就需要进行阻塞，阻塞是在acquireQueued方法中进行的 

<span style="font-size:14px;">    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;
                    failed = false;
                    return interrupted;
                }
                //阻塞，只有当前结点的前一个结点为SIGNAL时，才能当前线程阻塞</span><span style="font-size:14px;">
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }
</span>
队列有个特点就是头结点是占有锁的结点，如果获取锁成功，那么就需要将结点设置为头结点

还有一点要注意的是，这里是在循环中的，为了让当前线程运行的时候可以重新进行获取锁的操作

    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            return true;
        if (ws > 0) {
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }
    private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this);
        return Thread.interrupted();
    }
parkAndCheckInterrupt方法进行阻塞，只有当前结点的前一个结点为SIGNAL时，才能当前线程阻塞

上面是非公平锁的方式，公平锁的方式差不多，其lock方法如下

    final void lock() {
        acquire(1);
    }
没有直接设值的操作，都是进队列
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;
    }
    //如果当前线程前面有排队的线程则为true，如果当前线程前面没有排队线程或者线程为空则为false
    public final boolean hasQueuedPredecessors() {
        Node t = tail; 
        Node h = head;
        Node s;
        return h != t &&
            ((s = h.next) == null || s.thread != Thread.currentThread());
    }
这里主要是多了个hasQueuedPredecessors方法的判断，其他一样

公平锁主要是按照lock的顺序排队进行的，所以在队列前面有结点的时候就不尝试获取锁了

lock方法就讲完了，总结一下，就 3点：

1.尝试获取锁

2.失败就创建一个新结点放入队列尾部

3.阻塞当前的线程


接下来看下释放锁的过程，入口为unlock方法，其调用的是AQS的release方法：

    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) {
        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;
    }
    private void unparkSuccessor(Node node) {//node等于head
        int ws = node.waitStatus;
        if (ws < 0)//如果等待状态小于0则设置为0，代表释放锁
            compareAndSetWaitStatus(node, ws, 0);

        Node s = node.next;//获取头结点的下一个节点
        //从往前遍历找到最前面的waitStatus<=0的结点
        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);//启动头结点的下一个结点
    }

可以看到tryRelease有个c==0的判断，由于ReentrantLock是可重入锁，所以在多次lock操作的时候state就增加了多次，当lock和unlock成对出现时，当前线程才算没有占有锁

然后unparkSuccessor方法进行启动线程，这里一般是头结点后面的那个结点启动，如果找不到则找到最前面的等待状态小于0的结点进行启动


还记得lock的时候，线程在哪里停住了吗?是在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; 
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }
当线程重新启动的时候，循环继续，然后重新获取锁，如果成功了，那么才将该结点设置为头结点

这里设置头结点的时候没有使用线程安全的方法是因为能进来这里的只会有一个线程


ReentrantLock和AbstractQueuedSynchronizer就分析到这里，AbstractQueuedSynchronizer还是很复杂很强大的，限于能力，不能完全的理解透彻，只把自己理解的记录一下，如果后面有更深的理解会继续修改文章


参考文章：

http://ifeve.com/jdk1-8-abstractqueuedsynchronizer/

http://ifeve.com/introduce-abstractqueuedsynchronizer/




标签：

相关文章