ReentrantLock源码解读（lock、unlock原理）

不念不忘少年蓝@ 2021-06-22 15:38 513阅读 0赞

#### ReentrantLock简介 ####

ReentrantLock是JDK中JUC包下的工具类，顾名思义，它是可重入的，同一个线程可以多次获取锁。借助它可以实现只有一个线程同时访问临界区，解决并发问题。  
推荐的使用方式：

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

#### 实现例子 ####

public class ReentrantLockTest { 
    
    	private final ReentrantLock lock = new ReentrantLock();
    
    	private int num;
    
    	/** * 将num增加count * @param count */
    	public void addCount(int count) { 
    		lock.lock();
    		try { 
    			for (int i = 0; i < count; i++) { 
    				num++;
    			}
    		} finally { 
    			lock.unlock();
    		}
    	}
    
    	public static void main(String[] args) throws InterruptedException { 
    		ReentrantLockTest reentrantLockTest = new ReentrantLockTest();
    		Thread thread1 = new Thread(()->reentrantLockTest.addCount(200000));
    		Thread thread2 =new Thread(()->reentrantLockTest.addCount(200000));
    		thread1.start();
    		thread2.start();
    		thread1.join();
    		thread2.join();
    		System.out.println(reentrantLockTest.num);
    	}
    }

本例中，num是共享变量，多个线程都可以同时访问该变量并自增1，如果不进行控制，那么结果将不等于400000。例子中采用了ReentrantLock进行加锁的方式，保护了临界区，使得任意瞬间只有一个线程访问num，解决了并发问题。

#### 它是如何实现的呢 ####

先看ReentrantLock的lock方法：

/** * Acquires the lock. * * <p>Acquires the lock if it is not held by another thread and returns * immediately, setting the lock hold count to one. * * <p>If the current thread already holds the lock then the hold * count is incremented by one and the method returns immediately. * * <p>If the lock is held by another thread then the * current thread becomes disabled for thread scheduling * purposes and lies dormant until the lock has been acquired, * at which time the lock hold count is set to one. */
        public void lock() { 
            sync.lock();
        }

它调用了ReentrantLock的内部抽象类Sync的抽象方法lock()。Sync有两个实现NonfairSync和FairSync，可以实现公平与非公平。默认是非公平的。  
看一下NonfairSync的实现：

final void lock() { 
            	// 竞争锁，同时只有一个线程能成功
                if (compareAndSetState(0, 1))
                    setExclusiveOwnerThread(Thread.currentThread());
                else
                    acquire(1);
            }

首先CAS尝试了一把，在无竞争的情况下就直接获取到锁了。例子中就是thread1和thread2同时竞争，获取到锁的线程就将exclusiveOwnerThread置为当前线程。  
没获取到锁的线程则执行acquire(1);

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

再尝试去获取一次锁tryAcquire，它是AbstractQueuedSynchronizer的方法，在ReentrantLock进行了重写，并由nonfairTryAcquire实现：

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

在nonfairTryAcquire中我们看到，线程还有一次机会去获取锁，并增加了线程重入的实现，如果都失败，就返回false。  
当tryAcquire返回false，（线程还没有获取到锁）就要开始排队了。执行下面的acquireQueued(addWaiter(Node.EXCLUSIVE), arg)方法。

addWaiter(Node.EXCLUSIVE) //为线程创建一个排队的节点，类似于包装了一层

acquireQueued(final Node node, int arg) // 就是将排队的节点 加入到队列中

看到这里，我们有点明白了。lock方法首先会让线程去竞争锁，获取到锁的线程不受影响，继续执行。没获取到锁的线程就不好受了，需要排队了。  
接下来我们看看线程如何排队：

private Node addWaiter(Node mode) { 
            Node node = new Node(Thread.currentThread(), mode);
            // 先尝试一把直接将当前线程直接插入到队尾
            Node pred = tail;
            if (pred != null) { 
            	// 将tail设为node的前驱节点，这里是有并发问题的，但是没关系，多个node将tail设为前驱节点，还需要看tail是否承认node是它的后继
                node.prev = pred; 
                // 尝试让tail承认node是它的后继，成功的节点就成为后继了，成功加入队列
                if (compareAndSetTail(pred, node)) { 
                    pred.next = node;
                    return node;
                }
            }
            // 并发高的时候则会有线程加入队列失败，或tail为null（没有排队的人），走普通入队方式enq
            enq(node);
            return node;
        }

普通入队列方式：

private Node enq(final Node node) { 
        	// 一个显眼的自旋，意味着排不上队不走了
            for (;;) { 
                Node t = tail;
                if (t == null) {  // Must initialize
                	// 如果tail为空，说明你是第一个排队的。你们竞争一下谁排队头吧，没竞争上的线程下次循环tail就不为空了
                    if (compareAndSetHead(new Node()))
                        tail = head;
                } else { 
                	// 如果已经有人排队了，这个场景就和上面addWaiter中的场景一样了
                    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 (;;) { 
                	// 获取当前节点的前驱p
                    final Node p = node.predecessor();
                    // 如果node排在第一个（head不存线程），就试试再获取锁，获取不到则考虑挂起线程，线程阻塞。另外如果线程之前是挂起的，那么中断标志会为true，外层会设中断标记
                    if (p == head && tryAcquire(arg)) { 
                        setHead(node);
                        p.next = null; // help GC
                        failed = false;
                        return interrupted;
                    }
                    // 是否需要挂起线程，如果需要则挂起LockSupport.park(this)
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        interrupted = true;
                }
            } finally { 
                if (failed)
                    cancelAcquire(node);
            }
        }

看到这里，我们基本看到，没有拿到锁的线程是要被挂起了，需要等待被唤醒。  
挂起线程的方法采用LockSupport.park()方法：

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

*  接下来我看看看unLock方法：  
    看了lock方法，相信unlock方法将会简单一些：

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

调用了AbstractQueuedSynchronizer的release方法，与acquire相对，通俗理解就是将互斥量减去一定大小，这里是1：

public final boolean release(int arg) { 
        	// 尝试减小互斥量state，如果成功，唤醒队列中第一个线程
            if (tryRelease(arg)) { 
                Node h = head;
                if (h != null && h.waitStatus != 0)
                    unparkSuccessor(h);
                return true;
            }
            return false;
        }

这里我们主要看一下tryRelease和unparkSuccessor：  
tryRelease在ReentrantLock中被重写：

protected final boolean tryRelease(int releases) { 
                int c = getState() - releases;
                // 只能释放当前持有锁的线程
                if (Thread.currentThread() != getExclusiveOwnerThread())
                    throw new IllegalMonitorStateException();
                boolean free = false;
                // 如果state变为0，则认为释放锁成功。否则可能是重入多次，只释放了一层，还是返回false
                if (c == 0) { 
                    free = true;
                    setExclusiveOwnerThread(null);
                }
                setState(c);
                return free;
            }

private void unparkSuccessor(Node node) { 
            int ws = node.waitStatus;
            // 如果当前线程阻塞，则将WaitStatus设为0，0表示无状态
            if (ws < 0)
                compareAndSetWaitStatus(node, ws, 0);
    		
    		// 取到队列中第一个节点，如果s不为空，直接唤醒该线程。如果s为空或s的状态是canceled，就换一个节点进行唤醒
            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);
        }

回答上面的“这里为什么不从前往后找呢？”，大师是这么说的：  
/\*  
\* 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.  
\*/  
通常要unpark的线程就在下一个节点（这也是我们认为从前往后找的事实依据），但是这里只是一个判断，一般不会出现。但当出现后继是null的情况，我只能从后面找一个真正的后继。（在这里要是tail.prev莫名其妙为null了，那s就为null，就不做unpark了。）

#### 小结： ####

1、ReentrantLock采用了LockSupport.park、LockSupport.unpark来挂起线程和取消挂起。  
2、挂起的线程维护了一个双端队列，挂起的时候将节点加到队尾。取消挂起时从队头取一个节点进行unpark。  
3、ReentrantLock维护的互斥锁只能由一个线程持有，有AbstractOwnableSynchronizer中exclusiveOwnerThread变量存储。

本篇还没涉及到公平锁，condition等内容，后续会继续更新。