jdk源码解读并发包LockReentrantLock(1)lock()与unlock()方法走读

开发运维 2023-08-21 大白菜程序猿手机阅读

介绍

ReentrantLock 是一个互斥锁，在基本行为和机制上与synchonized一样，只不过synchonized用方法和声明访问了隐式的锁监视器，但是ReentrantLock 做了功能上的扩展。

当锁不被其他线程拥有，一个线程会成功的申请锁资源并立即返回。如果当前线程已经拥有了锁，再申请时也会立即返回。通过调用方法isHeldByCurrentThread()获取是否当前线程获得了锁，getHoldCount()得到获得几次锁资源。

此类的构造方法接受一个可选的公平参数。当设置为 true 时，在多个线程的争用下，这些锁倾向于将访问权授予等待时间最长的线程。否则此锁将无法保证任何特定访问顺序。与采用默认设置（使用不公平锁）相比，使用公平锁的程序在许多线程访问时表现为很低的总体吞吐量（即速度很慢，常常极其慢），但是在获得锁和保证锁分配的均衡性时差异较小。不过要注意的是，公平锁不能保证线程调度的公平性。因此，使用公平锁的众多线程中的一员可能获得多倍的成功机会，这种情况发生在其他活动线程没有被处理并且目前并未持有锁时。还要注意的是，未定时的 tryLock 方法并没有使用公平设置。因为即使其他线程正在等待，只要该锁是可用的，此方法就可以获得成功。

推荐使用用try-catch 块代码去调用lock(),如下：

 class X {
   private final ReentrantLock lock = new ReentrantLock();
   // ...
   public void m() {
     lock.lock();  // block until condition holds
     try {
       // ... method body
     } finally {
       lock.unlock()
     }
   }
 }}

除了实现lock接口，这个类还定义了一些public和protected方法去检查锁的状态。有些方法仅仅用来监控和维护。

这个类的序列化行为与内建的锁一样：反序列化的锁是没有获取锁状态，无论当它序列化时是否获取锁。

这个锁支持最大2147483647次的重入次数，超过这个数会报错。

类关系图：

从这个图可以看到ReentrantLock类实现了接口Lock和Serializable。

public class ReentrantLock implements Lock, java.io.Serializable {

ReentrantLock类的API调用都委托给一个内部类 Sync ,而该类继承AbstractQueuedSynchronizer类；

 	/**
 	 * Base of synchronization control for this lock. Subclassed
 	 * into fair and nonfair versions below. Uses AQS state to
 	 * represent the number of holds on the lock.
 	*/
        abstract static class Sync extends AbstractQueuedSynchronizer

而Sync又分为两个子类：公平锁和非公平锁，默认为非公平锁

 /**
	 * Sync object for fair locks
	 */
	static final class FairSync extends Sync

 //Sync object for Nonfair locks
 
 
static final class NonfairSync extends Sync

ReentrantLock调用lock()方法时的调用关系图

非公平锁类调用lock()方法时的调用关系：

代码解析：

nofairTryAcquire：

 /**
 * Performs non-fair tryLock.  tryAcquire is implemented in
 * subclasses, but both need nonfair try for trylock method.
 
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;
}

首先获取当前状态（初始化为0），当它等于0的时候，代表还没有任何线程获得该锁，然后通过CAS（底层是通过CompareAndSwapInt实现）改变state，如果设置成功设置当前线程为持有锁的线程；其他线程会直接返回false;当该线程重入的时候，state已经不等于0，这个时候并不需要CAS，因为该线程已经持有锁，然后会重新通过setState设置state的值，这里就实现了一个偏向锁的功能，即锁偏向该线程；

acquireQueued

只有当锁被一个线程持有，另外一个线程请求获得该锁的时候才会进入这个方法。

调用前首先调用addWaiter

addWaiter

 /**  * Creates and enqueues node for current thread and given mode.
*  * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared  
* @return the new node  
*/ 
 
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;         
if (compareAndSetTail(pred, node)) {
    pred.next = node;             
    return node;         
}     
}     
    enq(node);
    return node; 
}

首先，new一个节点，这个时候模式为：mode为 Node.EXCLUSIVE，默认为null即排它锁；

然后：

如果该队列已经有node即tail!=null，则将新节点的前驱节点置为tail，再通过CAS将tail指向当前节点，前驱节点的后继节点指向当前节点，然后返回当前节点；

如果队列为空或者CAS失败，则通过enq入队：

 /**
 * Inserts node into queue, initializing if necessary. See picture above.
 * @param node the node to insert
 * @return node's predecessor
 */
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;
            }
        }
    }
}

进队的时候，要么是第一个入队并且设置head节点并且循环设置tail，要么是add tail，如果CAS不成功，则会无限循环，直到设置成功，即使高并发的场景，也最终能够保证设置成功，然后返回包装好的node节点；

acquireQueued:

 /**
 * Acquires in exclusive uninterruptible mode for thread already in
 * queue. Used by condition wait methods as well as acquire.
 *
 * @param node the node
 * @param arg the acquire argument
 * @return {@code true} if interrupted while waiting
 */
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);
    }
}

该方法的主要作用就是真正让node入队，同时将已经进入虚拟队列的节点进行阻塞，我们看到，如果当前节点的前驱节点是head并且尝试获取锁的时候成功了，则直接返回，不需要阻塞；同时把当前节点设为头结点，原头结点则释放。

如果前驱节点不是头节点或者获取锁的时候失败了，则进行判定是否需要阻塞：

 /**
 * 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.
 *
 * @param pred node's predecessor holding status
 * @param node the node
 * @return {@code true} if thread should block
 */
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.
         */
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
/*
         * 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.
         */
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}

这段代码对该节点的前驱节点的状态进行判断，如果前驱节点已经处于signal状态，则返回true，表明当前节点可以进入阻塞状态；

否则，将前驱节点状态CAS置为signal状态，然后通过上层的for循环进入parkAndCheckInterrupt代码块park：

 /**
 * Convenience method to park and then check if interrupted
 *
 * @return {@code true} if interrupted
 */
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}

这个时候将该线程交给操作系统内核进行阻塞；

总体来讲，acquireQueued就是依靠前驱节点的状态来决定当前线程是否应该处于阻塞状态，如果前驱节点处于cancel状态，则丢弃这些节点，重新构建队列；

公平锁类调用lock()方法时的调用关系：

非公平锁类和公平锁类调用lock()时的区别：

非公平锁类调用lock（）时，不排队先尝试获取锁资源，修改状态，修改不成功再入队。具体实现先调用AbstractQueuedSynchronizer的方法
protected final boolean compareAndSetState(int expect, int update) ，而公平锁类是直接入队，不给插队的机会，当直接插队失败才会入队。

调用tryAcquire()时也不同，

公平锁类的tryAcquire()

 /**
 * Fair version of tryAcquire.  Don't grant access unless
 * recursive call or no waiters or is first.
 */
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;
}

非公平锁的tryAcquire()

 protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}

/** * Performs non-fair tryLock.  tryAcquire is implemented in * subclasses, but both need nonfair try for trylock method. */ 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

	class X {
	private final ReentrantLock lock = new ReentrantLock();
	// ...
	public void m() {
	lock.lock(); // block until condition holds
	try {
	// ... method body
	} finally {
	lock.unlock()
	}
	}
	}}

	/**
	* Base of synchronization control for this lock. Subclassed
	* into fair and nonfair versions below. Uses AQS state to
	* represent the number of holds on the lock.
	*/
	abstract static class Sync extends AbstractQueuedSynchronizer

	/**
	* Sync object for fair locks
	*/
	static final class FairSync extends Sync

	//Sync object for Nonfair locks


	static final class NonfairSync extends Sync

	/**
	* Performs non-fair tryLock. tryAcquire is implemented in
	* subclasses, but both need nonfair try for trylock method.

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

	/** * Creates and enqueues node for current thread and given mode.
	* * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared
	* @return the new node
	*/

	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;
	if (compareAndSetTail(pred, node)) {
	pred.next = node;
	return node;
	}
	}
	enq(node);
	return node;
	}

	/**
	* Inserts node into queue, initializing if necessary. See picture above.
	* @param node the node to insert
	* @return node's predecessor
	*/
	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;
	}
	}
	}
	}

	/**
	* Acquires in exclusive uninterruptible mode for thread already in
	* queue. Used by condition wait methods as well as acquire.
	*
	* @param node the node
	* @param arg the acquire argument
	* @return {@code true} if interrupted while waiting
	*/
	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);
	}
	}

	/**
	* 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.
	*
	* @param pred node's predecessor holding status
	* @param node the node
	* @return {@code true} if thread should block
	*/
	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.
	*/
	do {
	node.prev = pred = pred.prev;
	} while (pred.waitStatus > 0);
	pred.next = node;
	} else {
	/*
	* 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.
	*/
	compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
	}
	return false;
	}

	/**
	* Convenience method to park and then check if interrupted
	*
	* @return {@code true} if interrupted
	*/
	private final boolean parkAndCheckInterrupt() {
	LockSupport.park(this);
	return Thread.interrupted();
	}

	/**
	* Fair version of tryAcquire. Don't grant access unless
	* recursive call or no waiters or is first.
	*/
	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;
	}

	protected final boolean tryAcquire(int acquires) {
	return nonfairTryAcquire(acquires);
	}

jdk源码解读并发包LockReentrantLock(1)lock()与unlock()方法走读

介绍

类关系图：

ReentrantLock调用lock()方法时的调用关系图

非公平锁类调用lock()方法时的调用关系：

代码解析：

公平锁类调用lock()方法时的调用关系：

调用tryAcquire()时也不同，

如何在github上浏览文件

弹性布局如何设置最后一个元素的位置

Python与C++：哪个更适合初学者？

建造者模式实践

函数重写示例解析：实战案例中的应用精髓