亲宝软件园·资讯

展开

一文带你掌握Java ReentrantLock加解锁原理

阿笠在健身 人气:0

简要总结 ReentrantLock

实现原理:volatile 变量 + CAS设置值 + AQS + 两个队列

实现阻塞:同步队列 + CAS抢占标记为 valatile 的 state

实现等待唤醒:await :持有锁,park ->加入等待队列 ;signal:唤醒下一个等待队列节点,转移进入同步队列,然后CAS抢占或者按照阻塞队列等待抢占。接着 await 后续内容程序得以继续执行。

ReentrantLock 结构分析

ReentrantLock 继承了Lock接口, lock方法实际上是调用了Sync的子类NonfairSync(非公平锁)的lock方法。ReentrantLock的真正实现在他的两个内部类NonfairSync 和 FairSync中,默认实现是非公平锁。并且内部类都继承于内部类Sync,而Sync根本的实现则是大名鼎鼎的 AbstractQueuedSynchronizer 同步器(AQS)。

具体详见如下代码:

public class ReentrantLock implements Lock, java.io.Serializable {
    private static final long serialVersionUID = 7373984872572414699L;
    /** Synchronizer providing all implementation mechanics */
    private final Sync sync;
  
  public ReentrantLock() {
        sync = new NonfairSync();
    }
     abstract static class Sync extends AbstractQueuedSynchronizer {
       ……省略代码
     }
  
  //非公平锁
  static final class NonfairSync extends Sync {
        private static final long serialVersionUID = 7316153563782823691L;

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

        protected final boolean tryAcquire(int acquires) {
            return nonfairTryAcquire(acquires);
        }
    }
  // 公平锁
  static final class FairSync extends Sync {
        private static final long serialVersionUID = -3000897897090466540L;

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

        /**
         * Fair version of tryAcquire.  Don't grant access unless
         * recursive call or no waiters or is first.
         */
        protected final boolean tryAcquire(int acquires) {
……省略
        }
    }
  // lock 方法本质就是调用sync类
  public void lock() {
        sync.lock();
    }
}

lock 加锁过程

按照调用 lock 方法是否抢占锁成功,可以以调用 park 方法为界限,将加锁的过程分为两部分:一部分是当前线程被阻塞前,另一部分是线程被唤醒继续执行后。(这里以非公平锁为例)

阻塞前

1.直接通过CAS尝试获取锁,设置state为1。如果获取成功则将锁标识设为独占,就是是将当前线程设置给 exclusiveOwnerThread。

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

2.如果获取失败,再次尝试获取,调用acquire。

3.tryAcquire ->:判断锁是否被占有,如果空闲则再次尝试CAS获取锁;如果已被占有则对比占有锁的线程是否为本线程,是的话将state+1,这就是可重入锁的关键逻辑。

//AbstractQueuedSynchronizer
public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}
//ReentrantLock.NonfairSync
protected final boolean tryAcquire(int acquires) {
    return nonfairTryAcquire(acquires);
}
//ReentrantLock.Sync
final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
      // cas再次尝试获取
        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;
}

4.如果获取失败则将节点插入队列尾部,如果队列为空,则会初始化队列,并且设置头尾节点为空节点,再将Node设为尾节点。

// 获取锁失败
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))

// 加入同步队列
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;
      // 通过CAS设置尾节点为当前节点,前驱节点为之前的尾节点。
        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
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
          // 追加到队列尾
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

5.将新建的Node传入acquireQueued,获取前驱节点,如果节点就是head 头节点,那么尝试CAS竞争锁(head随时释放)。如果抢占成功将头节点设为自己。

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

6.如果没有抢占成功,则进入shouldParkAfterFailedAcquire逻辑,将前驱节点设置为Signal,表示后继节点(也就是当前节点)需要前驱节点去唤醒。设置完之后再次进入自旋锁,尝试获得锁。

关于Node的状态这里说明一下:

节点刚创建的时候,status=0,假设这时候本节点就是head节点,那么他会进入else逻辑,将自身状态设置为Signal,然后再次进入自旋,尝试获取锁。如果还是没有获取到锁,那么再次进入shouldParkAfterFailedAcquire方法后会进入第一个if逻辑,方法返回True。

/**
* Checks and updates status for a node that failed to acquire.
* Returns true if thread should block. This is the main signal
* control in all acquire loops.  Requires that pred == node.prev.
* 如果获取锁失败,检查并且更新节点。如果需要被park阻塞,返回true。
* 在所有的循环逻辑中,这是主要的信号控制逻辑。
*
* pred:表示前驱节点
* node:表示当前线程节点
*/
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.
         */
         // 若pred.waitStatus状态位大于0,说明这个前驱点已经取消了获取锁的操作,
         // doWhile循环会递归删除掉这些放弃获取锁的节点
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        /*
         * 节点刚创建的时候,status=0,逻辑会走到这里将自身状态设置为signal
         * 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.
         */
         //若状态位不为Node.SIGNAL,且没有取消操作,则会尝试将前驱节点状态位修改为Node.SIGNAL
        // 表示将会唤醒后继节点
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

7.第二次自旋获取失败后,由于前驱节点已经是Signal,这时进入parkAndCheckInterrupt,将当前线程阻塞,等待被唤醒。后续其他线程如果也尝试抢占锁,会同样被阻塞。

private final boolean parkAndCheckInterrupt() {
  // 阻塞线程
    LockSupport.park(this);
  // 线程继续执行
    return Thread.interrupted();
}

park方法被唤醒后

在其他线程释放锁资源后,唤醒下一个节点,park的后半部分逻辑继续执行。

1.继续执行之前Park之后的逻辑,在此处线程被唤醒。这里会返回中断标记,这也是为什么ReentrantLock可以相应中断的原因。

2.然后再次进入自旋锁,使用CAS获取到锁标记,将头节点设为当前节点,然后返回中断标记跳出循环。

3.至此,获取锁流程结束。

unlock 释放锁过程

1.尝试释放锁,用state减去1,判断是否等于0。如果等于0表示已经完全释放锁,将线程标记设为null。否则释放失败,表示当前线程仍在继续持有,继续持有说明有重入情况。

// ReentrantLock
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;
}
// 释放锁
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;
}

2.拿到头节点,然后解锁后继节点。如果当前节点状态小于0(signal=-1),则修改节点status为0。然后向后递归找到status小于等于0的节点(正常为0),调用unpark解除阻塞。返回解锁成功。

// 唤醒后继节点
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);
}

3.在这之后便继续开始执行之前被阻塞的线程中的逻辑。

到这里 ReentrantLock 的加解锁过程原理便讲解结束,关于条件队列的内容,有兴趣后续文章会做讲解。

对比 Synchronized

既然已经了解了 ReentrantLock ,那么在此对大家所熟知的 Synchronized 进行一个对比。

与Synchronized相同点

1.ReentrantLock和synchronized都是独占锁,只允许线程互斥的访问临界区。

但是实现上两者不同:synchronized加锁解锁的过程是隐式的,用户不用手动操作,优点是操作简单,但显得不够灵活。一般并发场景使用synchronized的就够了;ReentrantLock需要手动加锁和解锁,且解锁的操作尽量要放在finally代码块中,保证线程正确释放锁。ReentrantLock操作较为复杂,但是因为可以手动控制加锁和解锁过程,在复杂的并发场景中能派上用场。

2.ReentrantLock和synchronized都是可重入锁。

synchronized因为可重入因此可以放在被递归执行的方法上,且不用担心线程最后能否正确释放锁;而ReentrantLock在重入时要却确保重复获取锁的次数必须和重复释放锁的次数一样,否则可能导致其他线程无法获得该锁。

3.都可以实现线程之间的等待通知机制。使用synchronized结合Object上的wait和notify方法可以实现线程间的等待通知机制。ReentrantLock结合Condition接口同样可以实现这个功能。而且相比前者使用起来更清晰也更简单。

与Synchronized 不同点

加载全部内容

相关教程
猜你喜欢
用户评论