ReentrantLock可重入锁原理(硬核)

ReentrantLock可重入锁

文章目录

• ReentrantLock可重入锁
• • 1.继承关系
  • 2.非公平锁原理
  • • 2.1 加锁解锁流程
    • 2.2 加锁源码
    • 2.3 解锁源码
  • 3.可重入原理
  • 4.可打断原理
  • • 4.1 不可打断模式
    • 4.2 可打断模式
  • 5.公平锁原理
  • 6.条件变量原理
  • • 6.1 await流程
    • 6.2 signal流程
    • 6.3 源码

1.继承关系

2.非公平锁原理 2.1 加锁解锁流程

先从构造器开始看，默认为非公平锁实现

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

NonfairSync 继承自 AQS

当没有竞争(只有Thread-0)时:

当第一个竞争(Thread-1)出现时:

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

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

当前线程进入 acquireQueued 逻辑(维护AQS队列,创建节点Node且该节点关联当前线程)
1.acquireQueued 会在一个死循环中不断尝试获得锁，失败后进入 park 阻塞()
2.如果自己是紧邻着 head（排第二位），那么再次 tryAcquire 尝试获取锁，当然这时 state 仍为 1，失败
3.进入 shouldParkAfterFailedAcquire 逻辑，将前驱 node，即 head 的 waitStatus 改为 -1，这次返回 false

4.shouldParkAfterFailedAcquire 执行完毕回到 acquireQueued ，再次 tryAcquire 尝试获取锁，当然这时
state 仍为 1，失败

5.当再次进入 shouldParkAfterFailedAcquire 时，这时因为其前驱 node 的 waitStatus 已经是 -1，这次返回
true

6.进入 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 阻塞

2.2 加锁源码

//Sync继承自AQS
static final class NonfairSync extends Sync {
    private static final long serialVersionUID = 7316153563782823691L;

    //加锁实现
    final void lock() {
        // 首先用 cas 尝试（仅尝试一次）将 state 从 0 改为 1, 如果成功表示获得了独占锁
        if (compareAndSetState(0, 1))
            //cas成功,当前线程获得独占锁
            setExclusiveOwnerThread(Thread.currentThread());
        else
            // 如果尝试失败，进入[1]
            acquire(1);
    }
	
    //[1]AQS 继承过来的方法, 方便阅读, 放在此处
	public final void acquire(int arg) {
        //[2]tryAcquire
        if (!tryAcquire(arg) &&
            // 当 tryAcquire 返回为 false 时(真), 先调用 [4]addWaiter, 接着[5] acquireQueued
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }
    
    // [2]进入[3]
    protected final boolean tryAcquire(int acquires) {
        return nonfairTryAcquire(acquires);
    }
    
    //[4]AQS 继承过来的方法, 方便阅读, 放在此处
    private Node addWaiter(Node mode) {
        // 将当前线程关联到一个 Node 对象上, 模式为独占模式,Node.EXCLUSIVE
        Node node = new Node(Thread.currentThread(), mode);
        // 如果 tail 不为 null, cas 尝试将 Node 对象加入 AQS 队列尾部
        Node pred = tail;
        if (pred != null) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {
                // 双向链表
                pred.next = node;
                return node;
            }
        }
        // 尝试将 Node 加入 AQS, 进入[6]
        enq(node);
        return node;
    }
    
    //[6]AQS 继承过来的方法, 方便阅读, 放在此处
        private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            if (t == null) { // Must initialize
                // 还没有, 设置 head 为哨兵节点（不对应线程，状态为 0）
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                node.prev = t;
                // cas 尝试将 Node 对象加入 AQS 队列尾部
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }
    
    //[5]AQS 继承过来的方法, 方便阅读, 放在此处
    final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                // 上一个节点是 head, 表示轮到自己（当前线程对应的 node）了, 尝试获取 	
                if (p == head && tryAcquire(arg)) {
                    // 获取成功, 设置自己（当前线程对应的 node）为 head
                    setHead(node);
                    // 上一个节点 help GC
                    p.next = null; // help GC
                    failed = false;
                    // 返回中断标记 false
                    return interrupted;
                }
                if (
                    // 判断是否应当 park, 进入[7]
                    shouldParkAfterFailedAcquire(p, node) &&
                    // park 等待, 此时 Node 的状态被置为 Node.SIGNAL[方法八]
                    parkAndCheckInterrupt()
                   )
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }
    
    // [7]AQS 继承过来的方法, 方便阅读, 放在此处
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        // 获取上一个节点的状态
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            // 上一个节点都在阻塞, 那么自己也阻塞好了
            return true;
        // > 0 表示取消状态
        if (ws > 0) {
            // 上一个节点取消, 那么重构删除前面所有取消的节点, 返回到外层循环重试
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            // 这次还没有阻塞
			// 但下次如果重试不成功, 则需要阻塞，这时需要设置上一个节点状态为 Node.SIGNAL
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }
    
    //[8] 阻塞当前线程,AQS 继承过来的方法, 方便阅读, 放在此处
    private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this);
        return Thread.interrupted();
    }

}

2.3 解锁源码

    //解锁实现
    public void unlock() {
            sync.release(1);
    }

	// AQS 继承过来的方法, 方便阅读, 放在此处
    public final boolean release(int arg) {
        // 尝试释放锁, 进入 [1]
        if (tryRelease(arg)) {
            // 队列头节点 unpark
            Node h = head;
            if (
                // 队列不为 null
                h != null && 
                // waitStatus == Node.SIGNAL 才需要 unpark
                h.waitStatus != 0)
                // unpark AQS 中等待的线程, 进入[2]
                unparkSuccessor(h);
            return true;  //释放锁成功,返回true
        }
        return false;  //释放锁失败,返回false
    }


    //[1]Sync 继承过来的方法, 方便阅读, 放在此处
    protected final boolean tryRelease(int releases) {
        // state--
        int c = getState() - releases;
        if (Thread.currentThread() != getExclusiveOwnerThread())
            throw new IllegalMonitorStateException();
        //如果free=false表示释放锁失败
        boolean free = false;  
        // 支持锁重入, 只有 state 减为 0, 才释放成功
        if (c == 0) {
            free = true;   //free=true表示释放成功
            setExclusiveOwnerThread(null);  //将独占锁设为空
        }
        setState(c);  //修改独占所状态,如果释放成功则state=0
        return free;  //返回释放锁的状态
    }

    // [2] AQS 继承过来的方法, 方便阅读, 放在此处
    private void unparkSuccessor(Node node) {
        // 如果状态为 Node.SIGNAL 尝试重置状态为 0
        // 不成功也可以
    	int ws = node.waitStatus;
        if (ws < 0) {
        	compareAndSetWaitStatus(node, ws, 0);
        }
        // 找到需要 unpark 的节点, 但本节点从 AQS 队列中脱离, 是由唤醒节点完成的
        Node s = node.next;
        // 不考虑已取消的节点, 从 AQS 队列从后至前找到队列最前面需要 unpark 的节点
        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.可重入原理

static final class NonfairSync extends Sync {
	// ...
	// Sync 继承过来的方法, 方便阅读, 放在此处
	final boolean nonfairTryAcquire(int acquires) {
		final Thread current = Thread.currentThread();
        int c = getState();
		if (c == 0) {	//如果没有线程获得锁,不发生锁重入
			if (compareAndSetState(0, acquires)) { 
				setExclusiveOwnerThread(current);
				return true; //cas变更state状态成功后获取锁
			}
		}
        // 如果已经获得了锁, 线程还是当前线程, 表示发生了锁重入
		else if (current == getExclusiveOwnerThread()) {
        	// state++
        	int nextc = c + acquires;
        	if (nextc < 0) // overflow
        		throw new Error("Maximum lock count exceeded");
        	setState(nextc);
       	 	return true;
        }
        return false;
    }
    
    // Sync 继承过来的方法, 方便阅读, 放在此处
    protected final boolean tryRelease(int releases) {
    	// state--
    	int c = getState() - releases;
    	if (Thread.currentThread() != getExclusiveOwnerThread())
    		throw new IllegalMonitorStateException();
    	boolean free = false;
    	// 支持锁重入, 只有 state 减为 0, 
    	if (c == 0) {
    		free = true;
    		setExclusiveOwnerThread(null);
    	}
    	setState(c);
    	return free;
    }
}
    

4.可打断原理 4.1 不可打断模式

在此模式下，即使它被打断，仍会驻留在 AQS 队列中，一直要等到获得锁后方能得知自己被打断了 .

// Sync 继承自 AQS
static final class NonfairSync extends Sync {
    // ...
    
    private final boolean parkAndCheckInterrupt() {
    	// 如果打断标记已经是 true, 则 park 会失效
    	LockSupport.park(this);
    	// interrupted 会清除打断标记
    	return Thread.interrupted();
    }
    
    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()
                ) {
                    // 如果是因为 interrupt 被唤醒, 返回打断状态为 true
                    interrupted = true;
            	}
			}
		} finally {
            if (failed)
            cancelAcquire(node);
        }
	}
    
    public final void acquire(int arg) {
        if (
            !tryAcquire(arg) &&
			acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
			) {
        // 如果打断状态为 true
        selfInterrupt();
        }
	}
    
    static void selfInterrupt() {
        // 重新产生一次中断
        Thread.currentThread().interrupt();
    }
}

4.2 可打断模式

static final class NonfairSync extends Sync {
    public final void acquireInterruptibly(int arg) throws InterruptedException {
        if (Thread.interrupted())
        	throw new InterruptedException();
        // 如果没有获得到锁, 进入 ㈠
        if (!tryAcquire(arg))
        	doAcquireInterruptibly(arg);
    }
    
    // ㈠ 可打断的获取锁流程
    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()) {
                    // 在 park 过程中如果被 interrupt 会进入此
                    // 这时候抛出异常, 而不会再次进入 for (;;)
                    throw new InterruptedException();
                }
    		}
    	} finally {
            if (failed)
            cancelAcquire(node);
    	}
    }
    
}

5.公平锁原理

static final class FairSync extends Sync {
    private static final long serialVersionUID = -3000897897090466540L;
   
    final void lock() {
    	acquire(1);
    }
    
    // AQS 继承过来的方法, 方便阅读, 放在此处
    public final void acquire(int arg) {
        if (
        !tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
        ) {
        	selfInterrupt();  //
        }
   }
    
    // 与非公平锁主要区别在于 tryAcquire 方法的实现
    protected final boolean tryAcquire(int acquires) {
    	final Thread current = Thread.currentThread();  //获取当前线程
    	int c = getState();  //获取状态
    	if (c == 0) {
    		// 先检查 AQS 队列中是否有前驱节点, 没有才去竞争
    		if (!hasQueuedPredecessors() &&
    			compareAndSetState(0, acquires)) {
    				setExclusiveOwnerThread(current);
    				return true;
    		}
    	}
        //锁重入
    	else if (current == getExclusiveOwnerThread()) {
    		int nextc = c + acquires;  //c+1
    		if (nextc < 0)
    			throw new Error("Maximum lock count exceeded");
    		setState(nextc);
    		return true;
    	}
    	return false;
    }
    
    // ㈠ AQS 继承过来的方法, 方便阅读, 放在此处
    //检查前驱节点
    public final boolean hasQueuedPredecessors() {
        Node t = tail;
        Node h = head;
        Node s;
        // h != t 时表示队列中有 Node
        return h != t &&
            (
                // (s = h.next) == null 表示队列中还有没有老二
                (s = h.next) == null ||
                // 或者队列中老二线程不是此线程
                s.thread != Thread.currentThread()
            );
    }
    
}

6.条件变量原理

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

6.1 await流程

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

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

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

park 阻塞 Thread-0

6.2 signal流程

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

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

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

Thread-1 释放锁，进入 unlock 流程，略 .

6.3 源码

public class ConditionObject implements Condition, java.io.Serializable {
	private static final long serialVersionUID = 1173984872572414699L;
    // 第一个等待节点
    private transient Node firstWaiter;
    // 最后一个等待节点
    private transient Node lastWaiter;
    
    public ConditionObject() { }
    
    // ㈠ 添加一个 Node 至等待队列
    private Node addConditionWaiter() {
        Node t = lastWaiter;
        // 所有已取消的 Node 从队列链表删除, 见 ㈡
        if (t != null && t.waitStatus != Node.CONDITION) {
            unlinkCancelledWaiters();
            t = lastWaiter;
        }
        //创建一个关联当前线程的新 Node, 添加至队列尾部
        Node node = new Node(Thread.currentThread(), Node.CONDITION);
        if (t == null)
        	firstWaiter = node;
        else
        	t.nextWaiter = node;
        lastWaiter = node;
        return node;
   }
    
    // 唤醒 - 将没取消的第一个节点转移至 AQS 队列
    private void doSignal(Node first) {
        do {
            // 已经是尾节点了
            if ( (firstWaiter = first.nextWaiter) == null) {
           		 lastWaiter = null;
            }
       		 first.nextWaiter = null;
        } while (
            // 将等待队列中的 Node 转移至 AQS 队列, 不成功且还有节点则继续循环 ㈢
            !transferForSignal(first) &&
            // 队列还有节点
            (first = firstWaiter) != null
        );
    }
    
    // 外部类方法, 方便阅读, 放在此处
    // ㈢ 如果节点状态是取消, 返回 false 表示转移失败, 否则转移成功
    final boolean transferForSignal(Node node) {
        // 如果状态已经不是 Node.CONDITION, 说明被取消了
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        	return false;
        // 加入 AQS 队列尾部
        Node p = enq(node);
        int ws = p.waitStatus;
        if (
            // 上一个节点被取消
            ws > 0 ||
            // 上一个节点不能设置状态为 Node.SIGNAL
            !compareAndSetWaitStatus(p, ws, Node.SIGNAL)
        ) {
            // unpark 取消阻塞, 让线程重新同步状态
            LockSupport.unpark(node.thread);
        }
        return true;
    }
    
    // 全部唤醒 - 等待队列的所有节点转移至 AQS 队列
    private void doSignalAll(Node first) {
		lastWaiter = firstWaiter = null;
        do {
            Node next = first.nextWaiter;
            first.nextWaiter = null;
            transferForSignal(first);
            first = next;
        } while (first != null);
	}
    
    // ㈡
    private void unlinkCancelledWaiters() {
    	// ...
    }
    
    // 唤醒 - 必须持有锁才能唤醒, 因此 doSignal 内无需考虑加锁
    public final void signal() {
        if (!isHeldExclusively())
        	throw new IllegalMonitorStateException();
        Node first = firstWaiter;
        if (first != null)
        	doSignal(first);
    }
    
    // 全部唤醒 - 必须持有锁才能唤醒, 因此 doSignalAll 内无需考虑加锁
    public final void signalAll() {
        if (!isHeldExclusively())
        	throw new IllegalMonitorStateException();
        Node first = firstWaiter;
        if (first != null)
        	doSignalAll(first);
    }
    
    // 不可打断等待 - 直到被唤醒
    public final void awaitUninterruptibly() {
        // 添加一个 Node 至等待队列, 见 ㈠
        Node node = addConditionWaiter();
        // 释放节点持有的锁, 见 ㈣
        int savedState = fullyRelease(node);
        boolean interrupted = false;
        // 如果该节点还没有转移至 AQS 队列, 阻塞
        while (!isOnSyncQueue(node)) {
            // park 阻塞
            LockSupport.park(this);
            // 如果被打断, 仅设置打断状态
            if (Thread.interrupted())
            	interrupted = true;
         }
        // 唤醒后, 尝试竞争锁, 如果失败进入 AQS 队列
        if (acquireQueued(node, savedState) || interrupted)
        selfInterrupt();
    }
    
    // 外部类方法, 方便阅读, 放在此处
	// ㈣ 因为某线程可能重入，需要将 state 全部释放
    final int fullyRelease(Node node) {
        boolean failed = true;
        try {
            int savedState = getState();
            if (release(savedState)) {
            	failed = false;
            	return savedState;
        	} else {
        	throw new IllegalMonitorStateException();
        	}
        } finally {
            if (failed)
            node.waitStatus = Node.CANCELLED;
        }
    }
    
    // 打断模式 - 在退出等待时重新设置打断状态
	private static final int REINTERRUPT = 1;
	// 打断模式 - 在退出等待时抛出异常
	private static final int THROW_IE = -1;
	// 判断打断模式
    private int checkInterruptWhileWaiting(Node node) {
    	return Thread.interrupted() ?
    		(transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) : 0;
    }
    
    // ㈤ 应用打断模式
    private void reportInterruptAfterWait(int interruptMode)
    	throws InterruptedException {
    	if (interruptMode == THROW_IE)
    		throw new InterruptedException();
    	else if (interruptMode == REINTERRUPT)
    		selfInterrupt();
    }
    
    // 等待 - 直到被唤醒或打断
    public final void await() throws InterruptedException {
        if (Thread.interrupted()) {
        throw new InterruptedException();
        }
        // 添加一个 Node 至等待队列, 见 ㈠
        Node node = addConditionWaiter();
        // 释放节点持有的锁
        int savedState = fullyRelease(node);
        int interruptMode = 0;
        // 如果该节点还没有转移至 AQS 队列, 阻塞
        while (!isOnSyncQueue(node)) {
            // park 阻塞
            LockSupport.park(this);
           // 如果被打断, 退出等待队列
			if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
				break;
		}
        // 退出等待队列后, 还需要获得 AQS 队列的锁
        if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
            interruptMode = REINTERRUPT;
        // 所有已取消的 Node 从队列链表删除, 见 ㈡
        if (node.nextWaiter != null)
            unlinkCancelledWaiters();
        // 应用打断模式, 见 ㈤
        if (interruptMode != 0)
            reportInterruptAfterWait(interruptMode);
   }
    
   // 等待 - 直到被唤醒或打断或超时
public final long awaitNanos(long nanosTimeout) throws InterruptedException {
if (Thread.interrupted()) {
throw new InterruptedException();
}
// 添加一个 Node 至等待队列, 见 ㈠
Node node = addConditionWaiter();
// 释放节点持有的锁
int savedState = fullyRelease(node);
// 获得最后期限
final long deadline = System.nanoTime() + nanosTimeout;
int interruptMode = 0;
// 如果该节点还没有转移至 AQS 队列, 阻塞
while (!isOnSyncQueue(node)) {
// 已超时, 退出等待队列
if (nanosTimeout <= 0L) {
transferAfterCancelledWait(node);
break;
}
// park 阻塞一定时间, spinForTimeoutThreshold 为 1000 ns
if (nanosTimeout >= spinForTimeoutThreshold)
LockSupport.parkNanos(this, nanosTimeout);
// 如果被打断, 退出等待队列
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
nanosTimeout = deadline - System.nanoTime();
}
// 退出等待队列后, 还需要获得 AQS 队列的锁
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
// 所有已取消的 Node 从队列链表删除, 见 ㈡
if (node.nextWaiter != null)
unlinkCancelledWaiters();
// 应用打断模式, 见 ㈤
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
return deadline - System.nanoTime();
}
    
    // 等待 - 直到被唤醒或打断或超时, 逻辑类似于 awaitNanos
public final boolean awaitUntil(Date deadline) throws InterruptedException {
// ...
}
// 等待 - 直到被唤醒或打断或超时, 逻辑类似于 awaitNanos
public final boolean await(long time, TimeUnit unit) throws InterruptedException {
// ...
}
// 工具方法 省略 ...
}