JUC回顾之-可重入的互斥锁ReentrantLock

1.什么是可重锁ReentrantLock？

   就是支持重新进入的锁，表示该锁能够支持一个线程对资源的重复加锁。底层实现原理主要是利用通过继承AQS来实现的，也是利用通过对volatile state的CAS操作+CLH队列来实现；

支持公平锁和非公平锁。

CAS：Compare and Swap 比较并交换。CAS的思想很简单:3个参数，一个当前内存值V、预期值A，即将更新的值B，当前仅当预期值A和内存值V相等的时候，将内存值V修改为B，否则什么都不做。该操作是一个原子操作被广泛的用于java的底层实现中，在java中，CAS主要是有sun.misc.Unsafe这个类通过JNI调用CPU底层指令实现；更多底层的思想参考狼哥的文章cas的底层原理：https://www.jianshu.com/p/fb6e91b013cc

CLH队列：也叫同步队列，是带头结点的双向非循环列表，是AQS的主要实现原理（结构如下图所示）

 

 

 

2.ReentrantLock分为公平锁和非公平锁：区别是在于获取锁的机制上是否公平。

　　(1)公平锁：公平的获取锁，也就是等待时间最长的线程最优获取到锁，ReentraantLock是基于同步队列AQS来管理获取锁的线程。

                     在公平的机制下，线程依次排队获取锁，先进入队列排队的线程，等到时间越长的线程最优获取到锁。

     (2)非公平锁：而在“非公平”的机制下，在锁是可获取状态时，不管自己是否在对头都会获取锁。

     (3) 公平锁和非公平锁的对比：1、公平锁用来解决“线程饥饿”的问题，即先进入CLH同步队列等待的线程，即同步队列中的头结点总是先获取到锁。而非公平锁                                                会出现 一个线程连续多次获取锁的情况，使得其他线程只能在同步队列中等待。

                                             2、经过测试，10个线程，每一个线程获取100 000次锁，通过vmstat统计运行时系统线程上下文切换的次数；公平锁总耗时                                                   为：5754ms，而费公平锁总耗时为61ms。总耗时：公平锁/费公平锁=94.3倍，总切换次数 ：公平锁/费公平锁=133倍。非                                                 公平锁的线程切换更少，保证了更大的吞吐量。

 

   （4）可重入锁的结构如下：

             

 

3.非公平锁获取的实现源码如下：按照代码的执行顺序

（1）调用lock方法

 /**
     * Sync object for non-fair locks
     */
    static final class NonfairSync extends Sync {
        private static final long serialVersionUID = 7316153563782823691L;

        /**
         * 1.Performs lock.  Try immediate barge, backing up to normal
         * acquire on failure.调用lock方法
         */
        final void lock() {
  //判断当前state是否为0，即没有被任何线程获取的状态，如果是，CAS更新为1，当前线程获取到了非公平锁 if (compareAndSetState(0, 1))
  //设置当前锁的持有者线程 setExclusiveOwnerThread(Thread.currentThread()); else acquire(1); } 
    }

（2）如果状态不是0，CAS更新状态放回false走acquire(1)方法：

在讲aqs的时候说过这个方法，这里不做重复：

    /**2 获取非公平锁
     * Acquires in exclusive mode, ignoring interrupts.  Implemented
     * by invoking at least once {@link #tryAcquire},
     * returning on success.  Otherwise the thread is queued, possibly
     * repeatedly blocking and unblocking, invoking {@link
     * #tryAcquire} until success.  This method can be used
     * to implement method {@link Lock#lock}.
     *
     * @param arg the acquire argument.  This value is conveyed to
     *        {@link #tryAcquire} but is otherwise uninterpreted and
     *        can represent anything you like.
     */
    public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

(3).调用静态内部类NonfairSync重写AQS的tryAcquire(1)方法:

 /**
     * Sync object for non-fair locks
     */
    static final class NonfairSync extends Sync {
        private static final long serialVersionUID = 7316153563782823691L;


　　　　　//3.调用静态内部类NonfairSync重写AQS的tryAcquire方法 protected final boolean tryAcquire(int acquires) {
            return nonfairTryAcquire(acquires);
        }
    }

(4) 走nonfairTryAcquire(1)非公平锁的实现方法：重点分析下这个方法

  /**
         * 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();
            //c == 0 说明锁没有被任何线程所拥有，则CAS设置锁的状态为acquires if (c == 0) {
                if (compareAndSetState(0, acquires)) {
                   //设置当前线程为锁的拥有者
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            //如果锁的持有者已经是当期线程，更新锁的状态，这个地方就是为什么可重入的原因，如果获取锁的线程再次请求，则将同步状态的值增加，并返回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;
        }

 4.获取公平锁的过程

（1）公平锁源码获取源码如下：

  /**
     * Sync object for fair locks
     */
    static final class FairSync extends Sync {
        private static final long serialVersionUID = -3000897897090466540L;
        //直接调用acquire(1)方法 final void lock() { acquire(1); } /**
         * 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) {
               //c==0表示锁没有被任何线程锁拥有，首先判断当前线程是否为CLH同步队列的第一个线程；是的话，获取该锁，设置锁的状态 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;
        }
    }

从源码可以看出，和nonfairTryAcquire(int acquires)比较唯一不同的是 ，判断条件多了hasQueuePredecessors()方法，加入了当前节点是否有前驱节点的判断。如果返回true，表示有线程比当前线程等待的时间长，需要等待前驱线程获取并释放锁之后才能获取锁。返回false，表示当前的线程所在的节点为队列（CLH队列）的对头，可以直接获取锁。

 

/**
     * Queries whether any threads have been waiting to acquire longer
     * than the current thread.
     *
     * <p>An invocation of this method is equivalent to (but may be
     * more efficient than):
     *  <pre> {@code
     * getFirstQueuedThread() != Thread.currentThread() &&
     * hasQueuedThreads()}</pre>
     *
     * <p>Note that because cancellations due to interrupts and
     * timeouts may occur at any time, a {@code true} return does not
     * guarantee that some other thread will acquire before the current
     * thread.  Likewise, it is possible for another thread to win a
     * race to enqueue after this method has returned {@code false},
     * due to the queue being empty.
     *
     * <p>This method is designed to be used by a fair synchronizer to
     * avoid <a href="AbstractQueuedSynchronizer#barging">barging</a>.
     * Such a synchronizer's {@link #tryAcquire} method should return
     * {@code false}, and its {@link #tryAcquireShared} method should
     * return a negative value, if this method returns {@code true}
     * (unless this is a reentrant acquire).  For example, the {@code
     * tryAcquire} method for a fair, reentrant, exclusive mode
     * synchronizer might look like this:
     *
     *  <pre> {@code
     * protected boolean tryAcquire(int arg) {
     *   if (isHeldExclusively()) {
     *     // A reentrant acquire; increment hold count
     *     return true;
     *   } else if (hasQueuedPredecessors()) {
     *     return false;
     *   } else {
     *     // try to acquire normally
     *   }
     * }}</pre>
     *
     * @return {@code true} if there is a queued thread preceding the
     *         current thread, and {@code false} if the current thread
     *         is at the head of the queue or the queue is empty
     * @since 1.7
     */
    public final boolean hasQueuedPredecessors() {
        // The correctness of this depends on head being initialized
        // before tail and on head.next being accurate if the current
        // thread is first in queue.
        Node t = tail; // Read fields in reverse initialization order
        Node h = head;
        Node s;
        return h != t &&
            ((s = h.next) == null || s.thread != Thread.currentThread());
    }

 

 

 

4.释放锁：

（1）首先调用ReetrantLock重写父类AQS的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 } is thrown.
     *
     * @throws IllegalMonitorStateException if the current thread does not
     *         hold this lock
     */
    public void unlock() {
        sync.release(1);
    }

（2）调用AQS里面实现的释放互斥锁的方法：首先进入tryRelease()方法来尝试释放当前线程持有的锁，如果成功的话，调用unparkSuccessor唤醒后继线程。

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

（3）进入tryRelease方法我们重点分析：重写父类AQS里面的模板方法，进行锁的释放：

 
protected final boolean tryRelease(int releases) {
　　　　　　　//c是本次释放锁之后的状态 int c = getState() - releases;
           // 如果当前线程不是锁的持有者线程，则抛出异常 if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
            boolean free = false;
            //c==0表示锁被当前线程已经彻底释放，则将占有同步状态的线程设置为Null，即锁变为可获取的状态，这个时候才能返回true，否则返回false if (c == 0) {
                free = true;
                setExclusiveOwnerThread(null);
            }
            //否则设置同步状态
            setState(c);
            return free;
        }

（4）释放锁成功后，即锁变为可获取的状态后，调用unparkSuccessor唤醒后继线程，进入unparkSuccessor的源码：

/**
     * Wakes up node's successor, if one exists.
     *
     * @param node the node
     */
    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.
　　　　　　获取当前节点的有效的后继节点，这的有效是指后继节点s不为null并且waitStatus是<=0的，
          既没有被取消的状态。无效的话，通过for循环遍历，一直找到一个有效的节点
　　　　　　　　　　　

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

 

常用的方法：

 

1.创建一个 ReentrantLock ，默认是“非公平锁”。
        ReentrantLock()
    2. 创建策略是fair的 ReentrantLock。fair为true表示是公平锁，fair为false表示是非公平锁。
        ReentrantLock(boolean fair)
    3. 查询当前线程保持此锁的次数。 
        int getHoldCount()
       4. 返回目前拥有此锁的线程，如果此锁不被任何线程拥有，则返回 null。
           protected Thread getOwner()
    5.返回目前拥有此锁的线程，如果此锁不被任何线程拥有，则返回 null。
        protected Thread getOwner()
    6.返回一个 collection，它包含可能正等待获取此锁的线程。
        protected Collection<Thread> getQueuedThreads()
     7.返回正等待获取此锁的线程估计数。
        int getQueueLength()
     8.返回一个 collection，它包含可能正在等待与此锁相关给定条件的那些线程。
        protected Collection<Thread> getWaitingThreads(Condition condition)
    9.返回等待与此锁相关的给定条件的线程估计数。
        int getWaitQueueLength(Condition condition)
    10.查询给定线程是否正在等待获取此锁。
        boolean hasQueuedThread(Thread thread)
     11.查询是否有些线程正在等待获取此锁。
        boolean hasQueuedThreads()
    12.查询是否有些线程正在等待与此锁有关的给定条件。
        boolean hasWaiters(Condition condition)
     13.如果是“公平锁”返回true，否则返回false。
        boolean isFair()
     14.查询当前线程是否保持此锁。
        boolean isHeldByCurrentThread()
     15.查询此锁是否由任意线程保持。
        boolean isLocked()
     16.获取锁。
        void lock()
     17.如果当前线程未被中断，则获取锁。
        void lockInterruptibly()
     18.返回用来与此 Lock 实例一起使用的 Condition 实例。
        Condition newCondition()
     19.仅在调用时锁未被另一个线程保持的情况下，才获取该锁。
        boolean tryLock()
     20.如果锁在给定等待时间内没有被另一个线程保持，且当前线程未被中断，则获取该锁。
        boolean tryLock(long timeout, TimeUnit unit)
     21.试图释放此锁。
        void unlock()

 

应用多个线程的计数器：结果为 20000。

package concurrentMy.Volatiles;

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 
 * (类型功能说明描述)
 *
 * <p>
 * 修改历史:                                            <br>  
 * 修改日期            修改人员       版本             修改内容<br>  
 * -------------------------------------------------<br>  
 * 2016年4月8日 下午6:07:36   user     1.0        初始化创建<br>
 * </p> 
 *
 * @author        Peng.Li 
 * @version        1.0  
 * @since        JDK1.7
 */
public class VolatiteLock implements Runnable{
    // 不能保证原子性，如果不加lock的话
    private volatile int inc = 0;
    Lock lock = new ReentrantLock();
    

    /**
     * 
     * 理解：高速缓存 - 主存
     * 通过ReentrantLock保证原子性：读主存，在高速缓存中计算得到+1后的值，写回主存
     * (方法说明描述) 
     *
     */
    public  void increase() {
        lock.lock();
        try {
            inc++;
        } catch (Exception e) {
            e.printStackTrace();
        }finally{
            lock.unlock();
        }

    }
    

    public void run() {
        for (int i = 0; i < 10000; i++) {
            increase();
        }

    }

    public static void main(String[] args) throws InterruptedException {

        VolatiteLock v = new VolatiteLock();
        // 线程1
        Thread t1 = new Thread(v);
        // 线程2
        Thread t2 = new Thread(v);
        t1.start();
        t2.start();

        // for(int i=0;i<100;i++){
        // System.out.println(i);
        // }

        System.out.println(Thread.activeCount() + Thread.currentThread().getId() + Thread.currentThread().getName());

        while (Thread.activeCount() > 1)
            // 保证前面的线程都执行完
            Thread.yield();
        
        //20000
        System.out.println(v.inc);
    }

}