ReentrantLock和AbstractQueuedSynchronizer的分析

从ReentrantLock到AQS的分析

ReentrantLock的主要方法：

1、lock()：获取锁，没获取到会阻塞(Sync定义的抽象方法)

2、unLock()：释放锁(ReentrantLock定义的方法)

3、tryLock()：内部调用nonfairTryAcquire(1)以非公平的方式尝试获取锁，不会阻塞

4、tryLock(long timeout, TimeUnit unit)：调用各自重写的tryAcquire()方法来尝试获取锁，也会到队列排队，超过等待时间没获取到会中断等待

5、newCondition()：设置条件锁，唤醒的时候可以根据条件来唤醒对应的锁

6、lockInterruptibly()：可中断锁；持有该锁的线程或者在等待该锁的线程可以被中断

ReentrantLock可以实现公共锁和非公平锁，内部类Sync继承AQS来实现主要功能；

AQS主要是通过CAS+volatile int state标识+双向链表构成的先进先出队列实现获取锁、释放锁和等待锁的过程，ReentrantLock也遵循这一核心；

ReentrantLock和AbstractQueuedSynchronizer的分析

ReentrantLock的非公平锁：

非公平锁的获取：state=0(没有线程拥有锁)，这时先尝试获取锁，没有获取到的话再进入队列排队

// 默认为非公平锁
public ReentrantLock() { sync = new ReentrantLock.NonfairSync(); }

static final class NonfairSync extends ReentrantLock.Sync {
    /**
     * Performs lock.  Try immediate barge, backing up to normal
     * acquire on failure.
     */
    final void lock() {
        // 与公平锁的不同之处，会先试着抢占锁，没抢到再排队等待
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            // 进入队列排队，等待获得锁
            acquire(1);
    }

    // 注意：子类重写了AQS的tryAcquire()方法，子类调用是调用自己的
    // 公平锁和非公平锁的tryAcquire()不一样
    protected final boolean tryAcquire(int acquires) {
        return nonfairTryAcquire(acquires);
    }
}

// 非公平锁的tryAcquire()方法
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;
}

NonfairSync

公平锁获取：state=0(没有线程拥有锁)，这时判断队列是否有线程在等待，如果有的话则直接进入队列排队

// 设置是公平锁还是非公平锁 public ReentrantLock(boolean fair) { sync = fair ? new ReentrantLock.FairSync() : new ReentrantLock.NonfairSync(); } static final class FairSync extends ReentrantLock.Sync { final void lock() { // 这里不会像非公平锁一样先尝试获取锁，而是队列中如果有等待线程，则直接进入队列排队 acquire(1); } // 公平锁的tryAcquire() protected final boolean tryAcquire(int acquires) { final Thread current = Thread.currentThread(); int c = getState(); if (c == 0) { // 公平锁在state=0(没有线程持有锁)的时候，会先去判断等待队列里面是否有线程再排队，使用hasQueuedPredecessors()来判断 // 如果有线程在排队则不能获取到锁 // compareAndSetState(0, acquires) CSA原子更新 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"); // 上面的compareAndSetState和setState的区别： // 上面线程还没有获得锁，无法保证线程安全，所以会调用unsafe.compareAndSwapInt()来保证state更新的原子性 // 这里的线程已经获的了锁，是线程安全的，可以直接修改state的值 setState(nextc); return true; } return false; } }

public void unlock() { sync.release(1); } // AQS中的release方法，释放锁 public final boolean release(int arg) { // 调用Sync的tryRelease() if (tryRelease(arg)) { Node h = head; // 如果h.waitStatus!=0，说明有后继节点； // 因为入队列的节点的ws默认初始化为0，如果唤醒时ws=0，说明没有节点更新前驱节点的ws，这说明没有线程入队列 if (h != null && h.waitStatus != 0) // 释放锁的同时，唤醒队列中最前面的waitStatus<=0的等待线程 unparkSuccessor(h); return true; } return false; } // Sync实现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; } // 释放锁的同时，唤醒后继节点 private void unparkSuccessor(Node node) { int ws = node.waitStatus; // 为什么要设置头节点的ws为0 if (ws < 0) compareAndSetWaitStatus(node, ws, 0); // 头节点的下一个节点，如果不是null并且ws<=0则可以直接唤醒 Node s = node.next; // 后继节点为什么会等于null? 插入节点的时候pred.next不是线程安全的，所以有可能为null，但是从尾部遍历一定是可以的 if (s == null || s.waitStatus > 0) { s = null; for (Node t = tail; t != null && t != node; t = t.prev) // 找到队列中第一个waitStatus<=0的节点，并唤醒；后继节点为什么会出现=0的状态？新加入的tail节点，还没更新状态 if (t.waitStatus <= 0) s = t; } if (s != null) // 唤醒在自旋过程中被挂起的节点，继续自旋获取锁 LockSupport.unpark(s.thread); }

// AQS的acquire() public final void acquire(int arg) { // 先调用自己的tryAcquire()尝试获取锁，成功获取的话则返回true，该方法结束 // 没获取到的话则执行后面的入队操作，到队列中等待锁的获取 if (!tryAcquire(arg) && acquireQueued(addWaiter(AbstractQueuedSynchronizer.Node.EXCLUSIVE), arg)) selfInterrupt(); }

// ReentrantLock为独占模式 private Node addWaiter(Node mode) { Node node = new Node(Thread.currentThread(), mode); Node pred = tail; // 如果队列中有线程，则尝试将当前线程加到队尾 if (pred != null) { node.prev = pred; // 通过CAS原子更新将当前线程插入到队尾(如果有线程更新了尾结点，则插入失败，调用enq()来插入) if (compareAndSetTail(pred, node)) { pred.next = node; return node; } } // 队列为空或者原子更新失败 enq(node); return node; } private Node enq(final Node node) { for (;;) { Node t = tail; if (t == null) { // Must initialize // 此时队列为空，初始化队列；队列通过双向链表来实现，head节点没有线程 if (compareAndSetHead(new Node())) tail = head; } else { // 此时队列不为空，通过CAS将线程插入队尾，如果插入成功则返回节点，失败则进入下一轮循环 // 这里没有保证线程安全，可能出现尾分叉 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; } // 没拿到锁，判断是否将当前节点挂起 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } } private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { int ws = pred.waitStatus; // 如果节点的前驱节点的ws==SIGNAL，则返回true，将当前节点挂起，不用再一直尝试获取锁； // 在前驱节点释放之后，这个节点会被唤醒，唤醒之后再自旋获取锁 if (ws == Node.SIGNAL) /* * This node has already set status asking a release to signal it, so it can safely park. */ return true; // ws>0表示该节点的线程取消排队，将该节点踢出队列 if (ws > 0) { do { node.prev = pred = pred.prev; } while (pred.waitStatus > 0); pred.next = node; } else { /* * waitStatus must be 0 or PROPAGATE. */ // 此时前驱节点的ws不为SIGNAL，本轮循环将前驱节点的ws设置为SIGNAL，在下一轮循环，可以将当前节点挂起 compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } return false; } private final boolean parkAndCheckInterrupt() { LockSupport.park(this); return Thread.interrupted(); }

ReentrantLock和AbstractQueuedSynchronizer的分析

相关推荐