java.util.concurrent.locks包中的ReentrantLock之非公平策略解析

Posted 2022-09-03 charging-for-ycp

简介：

ReentrantLock作为一个可重入互斥锁，具有与Synchronized隐式监视器相同的功能，除此之外，还有更强的扩展性。

如果一个线程调用lock()，如果该锁未被另外一个线程持有，则成功获取锁并返回；如果当前线程已经持有该锁，则直接返回。可以通过isHeldByCurrentThread() 和 getHoldCount（）查看该线程是否已持有该锁以及次数。

Public ReentrantLock(boolean fairness):此构造器，可以通过设置fairness=true来保证ReentrantLock的公平性策略：等待时间最长的线程优先获得锁。采用公平策略比采用默认策略（非公平策略）的lock，在程序吞吐量方面前者远远低于后者；在尝试获取锁的次数与避免饥饿性方面却有着较小的差异。原文描述：（Programs using fair locks accessed by many threads may display lower overall throughput (i.e., are slower; often much slower) than those using the default setting, but have smaller variances in times to obtain locks and guarantee lack of starvation）。除此之外，锁的公平性无法保证线程调度的公平性。因此，可能存在以下情况：其中一个线程多次成功获取该锁，导致其他线程无法获取该锁。需要注意的是，tryLock()方法，在原线程释放锁的时候，即使等待队列中存在其他线程，那么调用tryLock（）的线程依然有可能提前获得该锁。

当我们调用lock()时后，推荐使用tryfinally 来进行锁的释放。

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

ReenttrantLock实现了序列化接口：当其反序列化时，是释放锁状态的，忽略其序列化时的锁状态。

 

默认情况下，ReenttrantLock采用的非公平策略。由其无参构造函数可以看出。

public ReentrantLock() 
        sync = new NonfairSync();
    

 这里主要介绍lock()与unlock()两种方法。

lock()

1.当我们获取锁的时候，它没有被其他线程持有，则持锁的数量=1，并立即返回；

2.如果该锁已被当前线程持有，则在原先持锁的数量+1，并立即返回；

3.如果该锁被其他线程持有，则该线程则进行阻塞，直到成功获取锁，并将持锁的数量=1.

final void lock() 
　　　　　　　//第一步
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
                acquire(1);
        

acquire()

以独占锁模式进行获取，忽略中断（延后中断）。如果尝试获取锁失败，则将该线程包装成一个EXCLUSIVE的Node加入一个等待队列中，通过调用tryAcquire()反复的进行阻塞与解除阻塞，直至成功

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

 非公平Sync的tryAcquire()实现

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;
        

 获取失败，入队：首先初始化等待队列

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;
            
        
　　　　　　//head==tail==null,进行初始化队列
        enq(node);
        return node;
    

 初始化队列

 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;
                
            
        
    

 其次在等待队列中尝试获取锁。该方法只会在机器出现异常或者与其他控件发生交互出现异常，才会取消获取锁。

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

 shouldParkAfterFailedAcquire(Node pred, Node node)

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;
    

 阻塞线程

parkAndCheckInterrupt()

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

 阻塞线程后，当持有锁的线程释放锁的时候，会通知等待队列中队首Head的后继节点的线程，将其唤醒，重新获取锁。