二探ReentrantLock非公平锁的加解锁过程

Posted 2022-06-01 cj_eryue

加锁 

ReentrantLock reentrantLock = new ReentrantLock();
        reentrantLock.lock();
        //...
        reentrantLock.unlock();

①java.util.concurrent.locks.ReentrantLock#lock

public void lock() 
    //具体的sync有公平和非公平（默认）两种
    sync.lock();

②java.util.concurrent.locks.ReentrantLock.NonfairSync#lock

final void lock() 
            if (compareAndSetState(0, 1))//这里体现的非公平
                setExclusiveOwnerThread(Thread.currentThread());
            else
                acquire(1);
        

③java.util.concurrent.locks.AbstractQueuedSynchronizer#acquire

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

到这里已经变得复杂了，一个个来分析：

④java.util.concurrent.locks.ReentrantLock.NonfairSync#tryAcquire

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

⑤java.util.concurrent.locks.ReentrantLock.Sync#nonfairTryAcquire

final boolean nonfairTryAcquire(int acquires) 
	//获取当前线程
    final Thread current = Thread.currentThread();
    //获取状态
    int c = getState();
    //如果为0 ，说明没有线程竞争。通过cas+1，每次线程重入该锁都会+1，释放都会-1，为0时释放锁。如果cas设置成功，则可以预计其它线程cas都是失败的，也就认为当前线程得到了锁，作为Running线程
    if (c == 0) 
        if (compareAndSetState(0, acquires)) 
              setExclusiveOwnerThread(current);
              return true;
        
    
    //如果c!=0，但是自己已经拥有锁，则只是简单的执行+1，并修改status值，因为没有竞争，所以直接通过setStat修改，而非cas，也就是说这里实现了偏向锁的功能！
    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;

如果返回fasle，我们来看下步骤③中的acquireQueued(addWaiter(Node.EXCLUSIVE), arg) 方法：

addWaiter方法负责把当前无法获得所锁的线程包装成Node添加到CLH队尾，mode参数表示是独占锁还是共享锁：

static final class Node     
    /** Marker to indicate a node is waiting in shared mode */    
    static final Node SHARED = new Node();    
    /** Marker to indicate a node is waiting in exclusive mode */   
    static final Node EXCLUSIVE = null;


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;
    	//如果当前队尾已经存在，则使用cas把当前线程更新为Tail
        if (pred != null) 
            node.prev = pred;
            if (compareAndSetTail(pred, node)) 
                pred.next = node;
                return node;
            
        
    //如果当前Tail为null或者使用cas把当前线程更新为Tail失败，则通过enq方法继续设置Tail
        enq(node);
        return node;
    


private Node enq(final Node node)     
    //循环调用CAS,将当前线程追加到队尾（或设置队头），并返回包装后的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;            
                    
            
    

把线程包装成Node对象的主要原因为除了用Node供虚拟队列用外，还用Node包装了各种线程状态：

/** waitStatus value to indicate thread has cancelled */
        static final int CANCELLED =  1;
        /** waitStatus value to indicate successor's thread needs unparking */
        static final int SIGNAL    = -1;
        /** waitStatus value to indicate thread is waiting on condition */
        static final int CONDITION = -2;
        /**
         * waitStatus value to indicate the next acquireShared should
         * unconditionally propagate
         */
        static final int PROPAGATE = -3;

• SIGNAL(-1)：线程的后继线程正/已被阻塞，当该线程release或cancel时要唤新这个后继线程（unpark）。

• CANCELLED(1)：因为超时或中断，该线程已经被取消

• CONDITION(-2)：表明该线程被处于条件队列，就是因为调用了了Condition.wait而被阻塞

• PROPAGATE(-3)：传播共享锁

• 0：0代表无状态

接着看acquireQueued，acquireQueued的主要作用是把已经追加到队列的线程节点（addWaiter方法返回值）进行阻塞，但阻塞前又通过tryAccauire重试是否能获得锁，如果重试成功则无需阻塞了，直接返回：

final boolean acquireQueued(final Node node, int arg) 
        boolean failed = true;
        try 
            boolean interrupted = false;
            for (;;) 
                final Node p = node.predecessor();
                //这里是个无限循环，但是不会出现死循环，原因在于parkAndCheckInterrupt会把当前线程挂起，从而阻塞住线程的调用栈
                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);
        
    

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

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) 
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
          	//规则1:如果前继的节点状态为SIGNAL，表明当前节点需要unpark,则返回成功，此时会执行parkAndCheckInterrupt方法，阻塞线程
            return true;
        if (ws > 0) 
            //规则2：如果前继节点状态大于0（CANCELLED状态），说明前置节点已经被放弃，则回溯到一个非取消的前继节点，返回false,acquireQueued方法的无限循环将递归调用该方法，直至规则1返回true,导致线程阻塞
            do 
                node.prev = pred = pred.prev;
             while (pred.waitStatus > 0);
            pred.next = node;
         else 
            //规则3：如果前继节点状态为非SIGNAL、非CANCELLED，则设置前继的状态为SIGNAL,返回false后进入acquireQueued方法的无限循环,同规则2
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        
        return false;
    

总的来说，shouldParkAfterFailedAcquire就是靠前继节点判断当前线程是否应该被阻塞，如果前继节点处于CANCELLED状态，则顺便删除。

 

至此，锁住线程的逻辑已经分析完成，下面讨论下解锁的过程。

解锁

请求锁不成功的线程会在acquireQueued方法中的parkAndCheckInterrupt方法里挂起，之后的代码必须等线程被解锁才能执行，假如现在线程被解锁，则执行interrupted = true;，之后又进入acquireQueued的无限循环。

但是得到解锁的线程不一定能获得锁，必须通过调用tryAcquire重新竞争，因为锁是非公平的，有可能被新加入的线程获得，从而导致刚被唤醒的线程再次被阻塞，这里充分体现了非公平机制

①java.util.concurrent.locks.ReentrantLock#unlock

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

②java.util.concurrent.locks.AbstractQueuedSynchronizer#release

public final boolean release(int arg) 
        if (tryRelease(arg)) 
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        
        return false;
    

//第一个可以unpark的线程，一般来说head.next == head,head就是第一个线程，但是head.next可能被取消或置为null，因此最稳妥的办法就是从后往前找第一个可用线程
private void unparkSuccessor(Node node) 
        /*
         * If status is negative (i.e., possibly needing signal) try
         * to clear in anticipation of signalling.  It is OK if this
         * fails or if status is changed by waiting thread.
         */
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);

        /*
         * 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.
         */
        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);
    

如果释放锁成功，则唤醒队列的第一个线程Head

protected final boolean tryRelease(int releases) 
            int c = getState() - releases;
            if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
            boolean free = false;
            if (c == 0) 
                free = true;
                setExclusiveOwnerThread(null);
            
            setState(c);
            return free;
        

tryRelease作用很明确，如果线程多次锁定，则进行多次释放，直至status==0则真正释放锁，即将status设置为0，因为无竞争，所也没有使用CAS。

 Lock Vs Synchronized

     AbstractQueuedSynchronizer通过构造一个基于阻塞的CLH队列容纳所有的阻塞线程，而对该队列的操作均通过Lock-Free（CAS）操作，但对已经获得锁的线程而言，ReentrantLock实现了偏向锁的功能。
        Synchronized的底层也是一个基于CAS操作的等待队列，但JVM实现的更精致，把等待队列分为ContentionList和EntryList，目的是为了降低现成的出列速度，当然也实现了偏向锁，从数据结构来说二者设计没有本质区别，但Synchronized还实现了自旋锁，并针对不同的系统和硬件体系进行了优化，而Lock则完全依靠系统阻塞挂起等待线程。

        当然Lock比Synchronized更适合在应用层拓展，可以继承AbstractQueuedSynchronizer定义各种实现，比如实现读写锁（ReentrantReadWriteLock）、公平或不公平锁；同时，Lock对应的Condition也比wait/notify要方便、灵活的多！