JUC-ReentrantLock

才疏学浅的木子

发布于 2023-10-17 08:33:01

1420

发布于 2023-10-17 08:33:01

文章被收录于专栏：CSDN文章

AQS

学习ReentrantLock就不得不知道AQS，因为ReentrantLock就是基于了AQS对象的

特点

用 state 属性来表示资源的状态（分独占模式和共享模式），子类需要定义如何维护这个状态，控制如何获取锁和释放锁
getState - 获取 state 状态
setState - 设置 state 状态
compareAndSetState - cas 机制设置 state 状态
独占模式是只有一个线程能够访问资源，而共享模式可以允许多个线程访问资源
提供了基于 FIFO 的等待队列，类似于 Monitor 的 EntryList
条件变量来实现等待、唤醒机制，支持多个条件变量，类似于 Monitor 的 WaitSet

ReentrantLock

相对于 synchronized 它具备如下特点

可中断
可以设置超时时间
可以设置为公平锁
支持多个条件变量

与 synchronized 一样，都支持可重入

可重入

可重入是指同一个线程如果首次获得了这把锁，那么因为它是这把锁的拥有者，因此有权利再次获取这把锁如果是不可重入锁，那么第二次获得锁时，自己也会被锁挡住

条件变量

synchronized 中也有条件变量，就是我们讲原理时那个 waitSet 休息室，当条件不满足时进入 waitSet 等待 ReentrantLock 的条件变量比 synchronized 强大之处在于，它是支持多个条件变量的，这就好比

synchronized 是那些不满足条件的线程都在一间休息室等消息
而 ReentrantLock 支持多间休息室，有专门等烟的休息室、专门等早餐的休息室、唤醒时也是按休息室来唤醒

原理

非公平锁实现原理

加锁流程

先看构造器,默认为非公平锁

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

没有竞争时

第一个竞争出现时

Thread-1 执行了

CAS 尝试将 state 由 0 改为 1，结果失败
进入 tryAcquire 逻辑，这时 state 已经是1，结果仍然失败
接下来进入 addWaiter 逻辑，构造 Node 队列

图中黄色三角表示该 Node 的 waitStatus 状态，其中 0 为默认正常状态 Node 的创建是懒惰的其中第一个 Node 称为 Dummy（哑元）或哨兵，用来占位，并不关联线程

当前线程进入 acquireQueued 逻辑

acquireQueued 会在一个死循环中不断尝试获得锁，失败后进入 park 阻塞
如果自己是紧邻着 head（排第二位），那么再次 tryAcquire 尝试获取锁，当然这时 state 仍为 1，失败
进入 shouldParkAfterFailedAcquire 逻辑，将前驱 node，即 head 的 waitStatus 改为 -1，这次返回 false

shouldParkAfterFailedAcquire 执行完毕回到 acquireQueued ，再次 tryAcquire 尝试获取锁，当然这时 state 仍为 1，失败
当再次进入 shouldParkAfterFailedAcquire 时，这时因为其前驱 node 的 waitStatus 已经是 -1，这次返回 true
进入 parkAndCheckInterrupt， Thread-1 park（灰色表示）

再次有多个线程经历上述过程竞争失败，变成这个样子

Thread-0 释放锁，进入 tryRelease 流程，如果成功

设置 exclusiveOwnerThread 为 null
state = 0

当前队列不为 null，并且 head 的 waitStatus = -1，进入 unparkSuccessor 流程找到队列中离 head 最近的一个 Node（没取消的），unpark 恢复其运行，本例中即为 Thread-1 回到 Thread-1 的 acquireQueued 流程

如果加锁成功（没有竞争），会设置 exclusiveOwnerThread 为 Thread-1，state = 1 head 指向刚刚 Thread-1 所在的 Node，该 Node 清空 Thread 原本的 head 因为从链表断开，而可被垃圾回收

如果这时候有其它线程来竞争（非公平的体现），例如这时有 Thread-4 来了

如果不巧又被 Thread-4 占了先

Thread-4 被设置为 exclusiveOwnerThread，state = 1
Thread-1 再次进入 acquireQueued 流程，获取锁失败，重新进入 park 阻塞

加锁源码

 static final class NonfairSync extends Sync {
        private static final long serialVersionUID = 7316153563782823691L;
        // 加锁实现
        final void lock() {
        	// 首先用 cas 尝试（仅尝试一次）将 state 从 0 改为 1, 如果成功表示获得了独占锁
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
            	//如果失败
                acquire(1);
        }

    public final void acquire(int arg) {
        if (!tryAcquire(arg) && //进入tryAcquire
        	//当tryAcquire返回false时候，先调用addWaiter，再调用acquireQueued
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

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) {
				//这里直接cas体现了非公平不去检查AQS
                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;
        }

addWaiter逻辑

    private Node addWaiter(Node mode) {
        //将当前线程关联到一个Node对象,模式为独占模式
        Node node = new Node(Thread.currentThread(), mode);
        
        Node pred = tail;
        //如果tail不为空cas尝试将Node对象加入AQS队列尾部
        if (pred != null) {
            node.prev = pred;
            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) { 
            	//还没有设置head 为哨兵节点
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
            	//cas将Node加入AQS尾部
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

acquireQueued中的流程

	final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
            	//判断这个节点的上一个节点
                final Node p = node.predecessor();
				//上一个节点是head，去尝试一次
                if (p == head && tryAcquire(arg)) {
                    //获取成功设置自己为head
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    //返回中断标记
                    return interrupted;
                }
                //尝试失败 / 上一个节点不是head
                if (shouldParkAfterFailedAcquire(p, node) &&  // 判断是否需要等待
                    parkAndCheckInterrupt()) 
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }


	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 {
	        // 需要设置上一个节点的值为SIGNAL
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }


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

注意：是否需要unpark是由当前节点的前驱节点的waitStatus == Node.SIGNAL来决定，而不是本节点的waitStatus

解锁源码

    public void unlock() {
        sync.release(1);
    }
    
	public final boolean release(int arg) {
		//尝试释放锁
        if (tryRelease(arg)) {
			//队头节点
            Node h = head;
           
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

tryRelease

        protected final boolean tryRelease(int releases) {
        	//state减去释放的
            int c = getState() - releases;
            if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
            boolean free = false;
			//如果state为01则设置owner为null
            if (c == 0) {
                free = true;
                setExclusiveOwnerThread(null);
            }
            setState(c);
            return free;
        }

unparkSuccessor

private void unparkSuccessor(Node node) {
		
		
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);

        Node s = node.next;
        // 不考虑已取消的节点
        if (s == null || s.waitStatus > 0) {
            s = null;
            //从队尾先前找需要unpark的节点
            for (Node t = tail; t != null && t != node; t = t.prev)
                if (t.waitStatus <= 0)
                    s = t;
        }
        if (s != null)
        	//启动线程
            LockSupport.unpark(s.thread);
    }

可重入原理

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

可打断原理

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

公平锁实现原理

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

条件变量实现原理

每个条件变量对应一个等待队列，其实先类是ConditionObject

await

开始 Thread-0 持有锁，调用 await，进入 ConditionObject 的 addConditionWaiter 流程创建新的 Node 状态为 -2（Node.CONDITION），关联 Thread-0，加入等待队列尾部

接下来进入 AQS 的 fullyRelease 流程，释放同步器上的锁

unpark AQS 队列中的下一个节点，竞争锁，假设没有其他竞争线程，那么 Thread-1 竞争成功

park 阻塞 Thread-0

signal 流程

假设 Thread-1 要来唤醒 Thread-0

进入 ConditionObject 的 doSignal 流程，取得等待队列中第一个 Node，即 Thread-0 所在 Node

执行 transferForSignal 流程，将该 Node 加入 AQS 队列尾部，将 Thread-0 的 waitStatus 改为 0，Thread-3 的 waitStatus 改为 -1

addConditionWaiter

private Node addConditionWaiter() {
            Node t = lastWaiter;
            
            if (t != null && t.waitStatus != Node.CONDITION) {
            	//从队列中删除
                unlinkCancelledWaiters();
                t = lastWaiter;
            }
            //创建一个节点放入
            Node node = new Node(Thread.currentThread(), Node.CONDITION);
            if (t == null)
                firstWaiter = node;
            else
                t.nextWaiter = node;
            lastWaiter = node;
            return node;
        }

doSignal

		//将没取消的第一个节点转移到AQS队列
        private void doSignal(Node first) {
            do {
                if ( (firstWaiter = first.nextWaiter) == null)
                    lastWaiter = null;
                first.nextWaiter = null;
            } while (!transferForSignal(first) &&  //将队列中的Node转移到AQS队列，不成功就继续循环
                     (first = firstWaiter) != null);
        }

final boolean transferForSignal(Node node) {
		  
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            return false;
        
        //加入AQS队列尾部
        Node p = enq(node);
        int ws = p.waitStatus;
		
		
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
            
            LockSupport.unpark(node.thread);
            
        return true;
    }

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2023-05-04，如有侵权请联系 cloudcommunity@tencent.com 删除

thread