美文网首页多线程
jdk源码解读-并发包-Lock-ReentrantLock(1

jdk源码解读-并发包-Lock-ReentrantLock(1

作者: 陈阳001 | 来源:发表于2017-08-18 13:42 被阅读44次

    介绍:

    ReentrantLock 是一个互斥锁,在基本行为和机制上与synchonized一样,只不过synchonized用方法和声明访问了隐式的锁监视器,但是ReentrantLock 做了功能上的扩展。
    ReentrantLock 被最后一个成功lock,但是还没unlock的线程拥有。当锁不被其他线程拥有,一个线程会成功的申请锁资源并立即返回。如果当前线程已经拥有了锁,再申请时也会立即返回。通过调用方法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次的重入次数,超过这个数会报错。

    类关系图:

    444.png

    从这个图可以看到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()方法时的调用关系:

    555.png

    代码解析:

    1.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的值,这里就实现了一个偏向锁的功能,即锁偏向该线程;

    2.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()方法时的调用关系:

    666.png

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

    1.非公平锁类调用lock()时,不排队先尝试获取锁资源,修改状态,修改不成功再入队。具体实现先调用AbstractQueuedSynchronizer的方法

    protected final boolean compareAndSetState(int expect, int update) ,而公平锁类是直接入队,不给插队的机会,当直接插队失败才会入队。

    2.调用tryAcquire()时也不同,

    1.公平锁类的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;
    }
    

    2.非公平锁的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 < 0) // overflow
    throw new Error("Maximum lock count exceeded");
            setState(nextc);
    return true;
        }
    return false;
    }
    

    公平锁类先调用!hasQueuedPredecessors()检查此节点前面有没有非头节点的节点。这样保证了顺序的获得锁资源。非公平锁不调用!hasQueuedPredecessors()直接CAS
    再来看unlock():

    1.调用流程图:

    777.png

    1.Reentantlock方法unlock():

    /**
         * Attempts to release this lock.
         *
         * <p>If the current thread is the holder of this lock then the hold
         * count is decremented.  If the hold count is now zero then the lock
         * is released.  If the current thread is not the holder of this
         * lock then {@link IllegalMonitorStateException} is thrown.
         *
         * @throws IllegalMonitorStateException if the current thread does not
         *         hold this lock
         */
        public void unlock() {
    
        sync.release(1);
    }
    

    unlock()调用AbstractQueuedSynchronizer的release。

    2.AbstractQueuedSynchronizer的release(int arg):

    /**
     * Releases in exclusive mode.  Implemented by unblocking one or
     * more threads if {@link #tryRelease} returns true.
     * This method can be used to implement method {@link Lock#unlock}.
     *
     * @param arg the release argument.  This value is conveyed to
     *        {@link #tryRelease} but is otherwise uninterpreted and
     *        can represent anything you like.
     * @return the value returned from {@link #tryRelease}
     */
    public final boolean release(int arg) {
    if (tryRelease(arg)) {
            Node h = head;
    if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
    return true;
        }
    return false;
    }
    

    tryRelease(arg)为true则调用unparkSucessor(h),否则直接返回false。

    3.我们再来看tryRelease(arg):

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

    当释放锁的thread与当前获得锁的线程不一致时,抛出异常,参数releasese是要释放的重入锁的个数,c是释放后还剩几个。如果c==0则返回true。无论返回是true还是false都会更新state的值。state为零说明锁资源已经可以竞争了,非零说明锁资源还在某个线程没有释放。如果为零,下一步应该是唤醒一个线程,使这个线程获得竞争锁的权利。

    4.unparkSucessor(h):

    /**
     * Releases in exclusive mode.  Implemented by unblocking one or
     * more threads if {@link #tryRelease} returns true.
     * This method can be used to implement method {@link Lock#unlock}.
     *
     * @param arg the release argument.  This value is conveyed to
     *        {@link #tryRelease} but is otherwise uninterpreted and
     *        can represent anything you like.
     * @return the value returned from {@link #tryRelease}
     */
    public final boolean release(int arg) {
    if (tryRelease(arg)) {
            Node h = head;
    if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
    return true;
        }
    return false;
    }
    

    这个作用即:当头结点的状态小于0,则将头结点的状态CAS为0,然后通过链表获取下一个节点,如果下一个节点为null或者不符合要求的状态,则从队尾遍历整个链表,直到遍历到离head节点最近的一个节点并且等待状态符合预期,则将头结点的后继节点置为该节点;

    对刚刚筛出来的符合要求的节点唤醒,也就是该节点获得 争夺 锁的权利。

    最后让我们回到调用lock()时,线程被park的那段代码:

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

    也就是说for循环又可以继续跑了,去做一系列的判断,并尝试获得锁,上面已经讲了,不再说了。这样,线程从lock()到unlock()发生的事情都大体讲清楚了。下一节,会分析await()和singnal()。

    相关文章

      网友评论

        本文标题:jdk源码解读-并发包-Lock-ReentrantLock(1

        本文链接:https://www.haomeiwen.com/subject/veiorxtx.html