ReentrantLock源码解析

Love The Way You Lie 2022-12-02 15:15 93阅读 0赞

ReentrantLock lock = new ReentrantLock(false);
    lock.lock();
    System.out.println("do something");
    lock.unlock();

首先来介绍 AbstractQueuedSynchronizer，它是一个双向链表的结构，有头节点和尾结点。来看看它的一些属性

// 在ReentrantLock语境下代表重入次数（0代表没有被持有）
    private volatile int state;
    // 独占锁模式下持有锁的线程（继承自AbstractOwnableSynchronizer）
    private transient Thread exclusiveOwnerThread;
    // 头节点
    private transient volatile Node head;
    // 尾节点
    private transient volatile Node tail;

接着，AbstractQueuedSynchronizer 有一个内部类 ：Node

static final class Node { 
    	// 共享锁标记
    	static final Node SHARED = new Node();
    	// 独占锁
    	static final Node EXCLUSIVE = null;	
    
    	// 等待的状态，有以下几种
    	volatile int waitStatus;	
    	// 取消状态（线程超时或者中断）
    	static final int CANCELLED =  1;
    	// 被标记的状态（此节点后面的节点都被阻塞，所以当前节点结束时需要需要阻塞后面的线程）
    	static final int SIGNAL    = -1;
    	
    	// 当前节点代表的线程
    	volatile Thread thread;
    	// 独占锁中用不到
    	Node nextWaiter;
    
    	// 当前节点上一个节点
    	volatile Node prev;
    	// 当前节点下一个节点
    	volatile Node next;
    	// 获取前面的一个节点
    
    	final Node predecessor() throws NullPointerException { 
    		Node p = prev;
    		if (p == null)
    			throw new NullPointerException();
    		else
    			return p;
    	}
    
    	Node(Thread thread, Node mode) {      // Used by addWaiter
    		this.nextWaiter = mode;
    		this.thread = thread;
    	}
    }

来看看lock的实现：

final void lock() { 
            // 非公平锁的lock方法，首先使用cas将state修改为1，如果cas成功，则将持有锁的线程设置为当前线程。
    	if (compareAndSetState(0, 1))
    		setExclusiveOwnerThread(Thread.currentThread());
    	else
            // 如果cas失败，则去获取锁。
    		acquire(1);
    }

获取锁的方法：

public final void acquire(int arg) { 
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

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) { 
    		// 没有线程持有这个锁, 参数acquires为 1
    		if (compareAndSetState(0, acquires)) { 
    			// 设置锁被持有线程为当前线程
    			setExclusiveOwnerThread(current);
    			// 返回true，整个lock流程走完
    			return true;
    		}
    	}
    	// 当前线程是锁被持有的线程（重入）
    	else if (current == getExclusiveOwnerThread()) { 
    		// 加上重入次数
    		int nextc = c + acquires;
    		if (nextc < 0) // overflow
    			throw new Error("Maximum lock count exceeded");
    		// 设置state，并返回true,整个lock流程走完
    		setState(nextc);
    		return true;
    	}
    	// 没有获取到锁，也没有重入，获取锁失败 
    	return false;
    }

如果获取到锁，整个lock()流程结束，如果没有获取到锁，则继续执行 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) 方法，先来看addWaiter(Node.Exclusive)方法：

private Node addWaiter(Node mode) { 
    	// 创建一个Node
    	Node node = new Node(Thread.currentThread(), mode);
    	// 尾结点赋值pred
    	Node pred = tail;
    	if (pred != null) { 
    		// 尾结点有元素，将当前节点插入到队尾
    		// 设置当前节点的前节点是当前尾节点
    		node.prev = pred;
    		// cas设置尾节点
    		if (compareAndSetTail(pred, node)) { 
    		// 设置成功，才将当前尾节点的下一个节点设为当前线程节点
    			pred.next = node;
    			return node;
    		}
    	}
    	// 如果入队没有这么顺利，进入enq(node)方法
    	enq(node);
    	return node;
    }

private Node enq(final Node node) { 
    	// 这是一个死循环
    	for (;;) { 
    		// 获取尾节点
    		Node t = tail;
    		// 如果尾节点是null，说明链表是空的
    		if (t == null) {  // Must initialize
    			// 这里cas一个新的Node来作为头节点，
    			if (compareAndSetHead(new Node()))
    				// 这里可能会有多个线程在设置头节点，这种写法不会出现并发问题，此时链表只有一个新Node，继续循环
    				tail = head;
    		} else { 
    			// 尾节点不为null，表示至少链表中有一个节点，将当前节点的前节点设置成尾节点
    			node.prev = t;
    			// cas设置尾节点
    			if (compareAndSetTail(t, node)) { 
    				// 将node成功设置为尾节点之后，才将尾节点的后节点设置为当前节点，
    				t.next = node;
    				//跳出循环
    				return t;
    			}
    		}
    	}
    }

在 enq(Node node) 方法中，如果设置尾节点失败，则不断循环，找到最新的尾节点，然后当前节点入队。

addWaiter(Node node) 方法就走完了，主要作用就是将代表当前线程的节点放入队尾。

public final void acquire(int arg) { 
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

回到刚才的方法，addWaiter(Node node)返回当前Node，来看看 acquireQueued(Node node, int arg)方法：

final boolean acquireQueued(final Node node, int arg) { 
    
    	boolean failed = true;
    	try { 
    		boolean interrupted = false;
    		// 又是一个死循环
    		for (;;) { 
    			// 获取当前节点的前一个节点
    			final Node p = node.predecessor();
    			// 前节点是头节点,并且尝试获得锁成功，tryAcquire方法前面介绍过
    			if (p == head && tryAcquire(arg)) { 
    				// 将当前节点设置为头节点
    				setHead(node);
    				// 原头节点的后节点设置为null,等待被回收
    				p.next = null; // help GC
    				failed = false;
    				return interrupted;
    			}
    			// 获取锁失败，执行阻塞逻辑，node是当前节点，p是前节点
    			if (shouldParkAfterFailedAcquire(p, node) &&
    				parkAndCheckInterrupt())
    				interrupted = true;
    		}
    	} finally { 
    		if (failed)
    			cancelAcquire(node);
    	}
    }

来看看shouldParkAfterFailedAcquire(Node pred, Node node) 方法

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { 
    	// 获取前节点的状态
    	int ws = pred.waitStatus;
    	if (ws == Node.SIGNAL)
    		// 如果是 SIGNAL 状态，则返回true
    		return true;
    	if (ws > 0) { 
    		// 状态大于0，表示状态为CANCELLED，这样的节点是不参与等待获取锁的，所以使用循环来跳过当前节点前面的CANCELLED状态节点
    		do { 
    			node.prev = pred = pred.prev;
    		} while (pred.waitStatus > 0);
    		pred.next = node;
    	} else { 
    		// 将前节点状态设置为SIGNAL
    		compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    	}
    	// 这里返回false，是为了回到acquireQueued方法的循环中，直到前节点状态为SIGNAL
    	return false;
    }

直到当前节点的前节点的状态为SIGNAL，进入parkAndCheckInterrupt()方法

private final boolean parkAndCheckInterrupt() { 
    	// 阻塞当前线程
    	LockSupport.park(this);
    	// 等线程继续执行的时候，返回线程的中断状态并复位
    	return Thread.interrupted();
    }

回到刚才的方法，等待的线程继续执行的时候，会继续执行循环，直到获取到锁，退出循环。下面的代码贴的是重复的，不用往上面翻了。finally 块中，代码执行失败的时候，会执行cancelAcquire(Node node)方法。failed 变量是局部变量，不会产生并发安全性问题。

获取锁之后，会返回interrupted状态，parkAndCheckInterrupt()方法中，返回true则表示线程获得锁的时候是中断状态，Thread.interrupted()会重置为非中断状态，所以等会要手动还原线程的状态。

public final void acquire(int arg) { 
    	if (!tryAcquire(arg) &&
    		acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
    		// 如果acquireQueued方法返回的中断状态为true，则需要还原线程中断状态
    		selfInterrupt();
    }
    
    static void selfInterrupt() { 
    	// 设置当前线程中断状态
    	Thread.currentThread().interrupt();
    }

获取锁的流程就到这里了。

然后我们来看释放锁：释放锁要相对简单些。

public void unlock() { 
    	sync.release(1);
    }

release(int arg)是AQS的方法，tryRelease(int arg)是ReentrantLock的实现方法：

public final boolean release(int arg) { 
    	if (tryRelease(arg)) { 
    		// 锁空闲状态，唤醒后面的节点
    		Node h = head;
    		// 头节点不是null，表示队列不为空，值得唤醒后面节点
    		// h.waitStatus != 0 为了防止头节点还未阻塞,由前面的分析中得知,线程在阻塞自己前必须设置前驱结点的状态为SIGNAL,否则它不会阻塞自己。
    		if (h != null && h.waitStatus != 0)
    			// 唤醒节点h
    			unparkSuccessor(h);
    		return true;
    	}
    	return false;
    }
    
    protected final boolean tryRelease(int releases) { 
    	// 此时releases为1，getState() 在ReentrantLock中表示锁的重入次数,c表示该锁释放一次之后剩下的重入次数
    	int c = getState() - releases;
    	// 当前线程不是锁持有线程，那还咋释放。。
    	if (Thread.currentThread() != getExclusiveOwnerThread())
    		throw new IllegalMonitorStateException();
    	// 表示锁是否空闲
    	boolean free = false;
    	if (c == 0) { 
    		// 锁空闲状态
    		free = true;
    		// 设置持有线程为null
    		setExclusiveOwnerThread(null);
    	}
    	// 修改重入次数
    	setState(c);
    	// 返回锁空闲状态，继续看上面release(int arg)方法
    	return free;
    }
    
    protected final void setState(int newState) { 
    	state = newState;
    }

private void unparkSuccessor(Node node) { 
    	// 获取状态 
    	int ws = node.waitStatus;
    	if (ws < 0)
    		// 将小于0的SIGNAL状态改为0（初始化状态）
    		compareAndSetWaitStatus(node, ws, 0);
    	// 获取节点的下一个
    	Node s = node.next;
    	if (s == null || s.waitStatus > 0) { 
    	// waitStatus大于0，说明是CANCELLED 状态，则继续往后找
    		s = null;
    		// 这里使用从后往前找的方法，找到离node最近的合适节点
    		for (Node t = tail; t != null && t != node; t = t.prev)
    			if (t.waitStatus <= 0)
    				s = t;
    	}
    	// 如果找到的节点不是null，则去唤醒他
    	if (s != null)
    		LockSupport.unpark(s.thread);
    }
    
    public static void unpark(Thread thread) { 
    	if (thread != null)
    		UNSAFE.unpark(thread);
    }