JAVA 并发 编程系列 （二）锁 之 ReentrantLoc

java 中的锁

• Lock接口

使用ReentrantLock实现线程同步

public class Run {

    public static void main(String[] args) {

        Lock lock = new ReentrantLock();

        //lambda写法
        new Thread(() -> runMethod(lock), "thread1").start();
        new Thread(() -> runMethod(lock), "thread2").start();
        new Thread(() -> runMethod(lock), "thread3").start();
        new Thread(() -> runMethod(lock), "thread4").start();
        //常规写法
        new Thread(new Runnable() {
            @Override
            public void run() {
                runMethod(lock);
            }
        }, "thread5").start();
    }

    private static void runMethod(Lock lock) {
        lock.lock();
        for (int i = 1; i <= 5; i++) {
            System.out.println("ThreadName:" + Thread.currentThread().getName() + (" i=" + i));
        }
        System.out.println();
        lock.unlock();
    }
}

结果

ThreadName:thread1 i=1
ThreadName:thread1 i=2
ThreadName:thread1 i=3
ThreadName:thread1 i=4
ThreadName:thread1 i=5

ThreadName:thread2 i=1
ThreadName:thread2 i=2
ThreadName:thread2 i=3
ThreadName:thread2 i=4
ThreadName:thread2 i=5

ThreadName:thread3 i=1
ThreadName:thread3 i=2
ThreadName:thread3 i=3
ThreadName:thread3 i=4
ThreadName:thread3 i=5

ThreadName:thread4 i=1
ThreadName:thread4 i=2
ThreadName:thread4 i=3
ThreadName:thread4 i=4
ThreadName:thread4 i=5

ThreadName:thread5 i=1
ThreadName:thread5 i=2
ThreadName:thread5 i=3
ThreadName:thread5 i=4
ThreadName:thread5 i=5

使用Lock对象和Condition实现等待/通知实例

主要方法对比如下：

• （1）Object的wait()方法相当于Condition类中的await()方法；
• （2）Object的notify()方法相当于Condition类中的signal()方法；
• （3）Object的notifyAll()方法相当于Condition类中的signalAll()方法；

public class LockConditionDemo {

    private Lock lock = new ReentrantLock();
    private Condition conditionA = lock.newCondition();
    private Condition conditionB = lock.newCondition();

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

        LockConditionDemo demo = new LockConditionDemo();

        new Thread(() -> demo.await(demo.conditionA), "thread1_conditionA").start();
        new Thread(() -> demo.await(demo.conditionB), "thread2_conditionB").start();
        new Thread(() -> demo.signal(demo.conditionA), "thread3_conditionA").start();
        System.out.println("稍等5秒再通知其他的线程！");
        Thread.sleep(5000);
        new Thread(() -> demo.signal(demo.conditionB), "thread4_conditionB").start();

    }

    private void await(Condition condition) {
        try {
            lock.lock();
            System.out.println("开始等待await！ ThreadName：" + Thread.currentThread().getName());
            condition.await();
            System.out.println("等待await结束！ ThreadName：" + Thread.currentThread().getName());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    private void signal(Condition condition) {
        lock.lock();
        System.out.println("发送通知signal！ ThreadName：" + Thread.currentThread().getName());
        condition.signal();
        lock.unlock();
    }
}

结果：

开始等待await！ ThreadName：thread1_conditionA
开始等待await！ ThreadName：thread2_conditionB
稍等5秒再通知其他的线程！
发送通知signal！ ThreadName：thread3_conditionA
等待await结束！ ThreadName：thread1_conditionA
发送通知signal！ ThreadName：thread4_conditionB
等待await结束！ ThreadName：thread2_conditionB

公平锁和非公平锁

概念很好理解，公平锁表示线程获取锁的顺序是按照线程加锁的顺序来分配，即先进先出，那么他就是公平的；非公平是一种抢占机制，是随机获得锁，并不是先来的一定能先得到锁，结果就是不公平的。
ReentrantLock提供了一个构造方法，可以很简单的实现公平锁或非公平锁，源代码构造函数如下：

public ReentrantLock(boolean fair) {
   sync = fair ? new FairSync() : new NonfairSync();
}
参数：fair为true表示是公平锁，反之为非公平锁，这里不再写代码测试。

Sync为ReentrantLock里面的一个内部类，它继承AQS（AbstractQueuedSynchronizer），它有两个子类：公平锁FairSync和非公平锁NonfairSync。

ReentrantLock里面大部分的功能都是委托给Sync来实现的，同时Sync内部定义了lock()抽象方法由其子类去实现，默认实现了nonfairTryAcquire(int acquires)方法，可以看出它是非公平锁的默认实现方式。下面我们看非公平锁的

lock()方法：

    public void lock() {
        sync.lock();
    }
    
    final void lock() {
            //尝试获取锁
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
        //获取失败，调用AQS的acquire(int arg)方法
            acquire(1);
    }
    
    public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }
    
    
    protected final boolean tryAcquire(int acquires) {
        return nonfairTryAcquire(acquires);
    }
    
    
    final boolean nonfairTryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        //获取同步状态
        int c = getState();
        //state == 0,表示没有该锁处于空闲状态
        if (c == 0) {
            //获取锁成功，设置为当前线程所有
            if (compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        //判断是否重入 
        //判断锁持有的线程是否为当前线程
        else if (current == getExclusiveOwnerThread()) {
            
            int nextc = c + acquires; //getState()+1
            if (nextc < 0) // overflow
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false;
    }

unlock 方法

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

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


    protected final boolean tryRelease(int releases) {
    
        int c = getState() - releases;//getState() - 1
        //如果释放的不是持有锁的线程，抛出异常
        if (Thread.currentThread() != getExclusiveOwnerThread())
            throw new IllegalMonitorStateException();
        boolean free = false;
        //state == 0 表示已经释放完全了，其他线程可以获取同步状态了
        if (c == 0) {
            free = true;
            setExclusiveOwnerThread(null);
        }
        setState(c);
        return free;
    }
    
    
    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());
    }

比较非公平锁和公平锁获取同步状态的过程，会发现两者唯一的区别就在于公平锁在获取同步状态时多了一个限制条件：hasQueuedPredecessors()

该方法主要做一件事情：主要是判断当前线程是否位于CLH同步队列中的第一个。如果是则返回true，否则返回false。

ReentrantLock 的 公共锁

tryAcquire 方法：

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

ReentrantLock与synchronized的区别

前面提到ReentrantLock提供了比synchronized更加灵活和强大的锁机制，那么它的灵活和强大之处在哪里呢？他们之间又有什么相异之处呢？

首先他们肯定具有相同的功能和内存语义。

• 与synchronized相比，ReentrantLock提供了更多，更加全面的功能，具备更强的扩展性。例如：时间锁等候，可中断锁等候，锁投票。
• ReentrantLock还提供了条件Condition，对线程的等待、唤醒操作更加详细和灵活，所以在多个条件变量和高度竞争锁的地方，ReentrantLock更加适合（以后会阐述Condition）。
• ReentrantLock提供了可轮询的锁请求。它会尝试着去获取锁，如果成功则继续，否则可以等到下次运行时处理，而synchronized则一旦进入锁请求要么成功要么阻塞，所以相比synchronized而言，ReentrantLock会不容易产生死锁些。
• ReentrantLock支持更加灵活的同步代码块，但是使用synchronized时，只能在同一个synchronized块结构中获取和释放。注：ReentrantLock的锁释放一定要在finally中处理，否则可能会产生严重的后果。
  ReentrantLock支持中断处理，且性能较synchronized会好些。

[图片上传失败...(image-a5a35b-1513655992049)]

使用ReentrantReadWriteLock实现并发

• ReentrantReadWriteLock有两个锁：一个是与读相关的锁，称为“共享锁”；另一个是与写相关的锁，称为“排它锁”。也就是多个读锁之间不互斥，读锁与写锁互斥，写锁与写锁互斥。

public class ReentrantReadWriteLock implements ReadWriteLock , java.io.Serializable{
    
    /** 内部类  读锁 */
    private final ReentrantReadWriteLock.ReadLock readerLock;
    /** 内部类  写锁 */
    private final ReentrantReadWriteLock.WriteLock writerLock;

    /** 使用默认（非公平）的排序属性创建一个新的 ReentrantReadWriteLock */
    public ReentrantReadWriteLock() {
        this(false);
    }
    
    public ReentrantReadWriteLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
        readerLock = new ReadLock(this);
        writerLock = new WriteLock(this);
    }

    /** 返回用于写入操作的锁 */
    public ReentrantReadWriteLock.WriteLock writeLock() { return writerLock; }
    /** 返回用于读取操作的锁 */
    public ReentrantReadWriteLock.ReadLock  readLock()  { return readerLock; }
    
    
    .....

}

在ReentrantLock中使用一个int类型的state来表示同步状态，
该值表示锁被一个线程重复获取的次数。
但是读写锁ReentrantReadWriteLock内部维护着两个一对锁，需要用一个变量维护多种状态。所以读写锁采用“按位切割使用”的方式来维护这个变量，将其切分为两部分，高16为表示读，低16为表示写。
[图片上传失败...(image-fc73b3-1513655992049)]

假如当前同步状态为S，那么写状态等于 S & 0x0000FFFF（将高16位全部抹去），读状态等于S >>> 16(无符号补0右移16位)。代码如下：

    static final int SHARED_SHIFT   = 16;
    static final int SHARED_UNIT    = (1 << SHARED_SHIFT);
    static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1;
    static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;

    /** Returns the number of shared holds represented in count  */
    static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }
    /** Returns the number of exclusive holds represented in count  */
    static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }

写锁

写锁的获取最终会调用tryAcquire(int arg)，该方法在内部类Sync中实现：

    protected final boolean tryAcquire(int acquires) {

        Thread current = Thread.currentThread();
        //当前锁的个数
        int c = getState();
        //写锁
        int w = exclusiveCount(c);
        if (c != 0) {
            //c != 0 && w == 0 表示存在读锁
            //当前线程不是已经获取写锁的线程
            // (Note: if c != 0 and w == 0 then shared count != 0)
            if (w == 0 || current != getExclusiveOwnerThread())
                return false;
            //超出最大范围
            if (w + exclusiveCount(acquires) > MAX_COUNT)
                throw new Error("Maximum lock count exceeded");
            // Reentrant acquire
            setState(c + acquires);
            return true;
        }
        //是否需要阻塞
        //writerShouldBlock() 公平锁 会判断是否是头结点 但是非公平锁不会
        if (writerShouldBlock() || !compareAndSetState(c, c + acquires))
            return false;
        //设置获取锁的线程为当前线程
        setExclusiveOwnerThread(current);
        return true;
    }

• 读锁未被占用(AQS state高16位为0) ，写锁未被占用（state低16位为0）或者占用写锁的线程是当前线程
• writerShouldBlock()方法返回false,即不阻塞写线程
• 当前写锁占有量小于最大值(2^16 -1)，否则抛出Error("Maximum lock count exceeded")
• 通过CAS竞争将写锁状态+1(将state低16位同步+1)

条件1使得写锁与读锁互斥，ReentrantReadWriteLock并没有读锁升级的功能。

写锁释放

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

写锁释放

读锁

    public void lock() {
        sync.acquireShared(1);
    }

    public final void acquireShared(int arg) {
        if (tryAcquireShared(arg) < 0)
            doAcquireShared(arg);
    }
    
    
    protected final int tryAcquireShared(int unused) {
        /*
         * Walkthrough:
         * 1. If write lock held by another thread, fail.
         * 2. Otherwise, this thread is eligible for
         *    lock wrt state, so ask if it should block
         *    because of queue policy. If not, try
         *    to grant by CASing state and updating count.
         *    Note that step does not check for reentrant
         *    acquires, which is postponed to full version
         *    to avoid having to check hold count in
         *    the more typical non-reentrant case.
         * 3. If step 2 fails either because thread
         *    apparently not eligible or CAS fails or count
         *    saturated, chain to version with full retry loop.
         */
     //当前线程
        Thread current = Thread.currentThread();
        int c = getState();
        
        //exclusiveCount(c)计算写锁
        //如果存在写锁，且锁的持有者不是当前线程，直接返回-1
        //存在锁降级问题，后续阐述
        if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current)
            return -1;
        
         //读锁
        int r = sharedCount(c);
        
        
        
        /*
         * readerShouldBlock():读锁是否需要等待（公平锁原则）
         * r < MAX_COUNT：持有线程小于最大数（65535）
         * compareAndSetState(c, c + SHARED_UNIT)：设置读取锁状态
         */
        if (!readerShouldBlock() &&
            r < MAX_COUNT &&
            compareAndSetState(c, c + SHARED_UNIT)) {
            if (r == 0) {//如果r=0, 表示，当前线程为第一个获取读锁的线程

                firstReader = current;
                firstReaderHoldCount = 1;
            } else if (firstReader == current) {
            //如果第一个获取读锁的对象为当前对象，
            //将firstReaderHoldCount 加一(重入)
                firstReaderHoldCount++;
            } else {
            //成功获取锁后，如果不是第一个获取多锁的线程，
            //将该线程持有锁的次数信息，放入线程本地变量中，
            //方便在整个请求上下文（请求锁、释放锁等过程中）使用持有锁次数信息。
            
            
                HoldCounter rh = cachedHoldCounter;//   获取最后一次获取到读状态的线程
                
                //rh == null(当前线程是第二个获取的),
                //或者当前线程和rh不是同一个，那么获取到当前线程的HoldCounter
                if (rh == null || rh.tid != getThreadId(current))
                    cachedHoldCounter = rh = readHolds.get();
                    
            //如果rh就是当前线程的HoldCounter
            //并且当前线程获取到的读状态位0那么给当前线程的HoldCounter设置为rh
                else if (rh.count == 0)
                    readHolds.set(rh);
            //获取到的读锁数加1
                rh.count++;
            }
            return 1;
        }
        return fullTryAcquireShared(current);
    }
    
    //读锁的获取比较繁琐，但是总的来说还是通过CAS设置同步状态
    final int fullTryAcquireShared(Thread current) {
        /*
         * This code is in part redundant with that in
         * tryAcquireShared but is simpler overall by not
         * complicating tryAcquireShared with interactions between
         * retries and lazily reading hold counts.
         */
        HoldCounter rh = null;
        for (;;) {
            int c = getState();
            //如果已经有写锁被获取
            if (exclusiveCount(c) != 0) {
                if (getExclusiveOwnerThread() != current)
                    return -1;
                // else we hold the exclusive lock; blocking here
                // would cause deadlock.
        
            //如果获取写锁的线程是当前线程则继续保持这个写锁
            //如果此时应该进入阻塞
            } else if (readerShouldBlock()) {
                // Make sure we're not acquiring read lock reentrantly
                //确保我们没有重新获得读锁
                if (firstReader == current) {
                    // assert firstReaderHoldCount > 0;
                } else {
                    if (rh == null) {
                        rh = cachedHoldCounter;
                        if (rh == null || rh.tid != getThreadId(current)) {
                            rh = readHolds.get();
                            if (rh.count == 0)
                    //如果当前线程的读锁为0就remove，因为后面会set
                                readHolds.remove();
                        }
                    }
                    if (rh.count == 0)
                        return -1;
                }
            }
        //尝试CAS设置同步状态
        //后续操作和tryAquireShared基本一致
            if (sharedCount(c) == MAX_COUNT)
                throw new Error("Maximum lock count exceeded");
            if (compareAndSetState(c, c + SHARED_UNIT)) {
                if (sharedCount(c) == 0) {
                    firstReader = current;
                    firstReaderHoldCount = 1;
                } else if (firstReader == current) {
                    firstReaderHoldCount++;
                } else {
                    if (rh == null)
                        rh = cachedHoldCounter;
                    if (rh == null || rh.tid != getThreadId(current))
                        rh = readHolds.get();
                    else if (rh.count == 0)
                        readHolds.set(rh);
                    rh.count++;
                    cachedHoldCounter = rh; // cache for release
                }
                return 1;
            }
        }
    }

读锁获取

读锁的获取条件要满足:

• 当前的写锁未被占有(AQS state变量低16位为0) 或者当前线程是写锁占有的线程
• readerShouldBlock()方法返回false
• 当前读锁占有量小于最大值(2^16 -1)
• 成功通过CAS操作将读锁占有量+1(AQS的state高16位同步加1)

条件1使得读锁与写锁互斥，除非当前申请读操作的线程是占有写锁的线程，即实现了写锁降级为读锁。

条件2在非公平模式下执行的是NonfairSync类的readerShouldBlock()方法:

    final boolean readerShouldBlock() {
        return apparentlyFirstQueuedIsExclusive();
    }
    final boolean apparentlyFirstQueuedIsExclusive() {
        Node h, s;
        return (h = head) != null &&
            (s = h.next)  != null &&
            !s.isShared()         &&
            s.thread != null;
    }

如果AQS的锁等待队列head节点后的节点非共享节点（等待读锁的节点），将返回true。（即申请读锁失败）

条件2在公平模式下执行的是FairSync类的readerShouldBlock方法:

    final boolean readerShouldBlock() {
        return hasQueuedPredecessors();
    }
    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());
    }

只要AQS锁等待队列的头尾不为空，并且存在head后的节点并且节点的线程非当前线程，返回true。(申请的节点必须是 最前面那个节点 才能申请成功)

读锁释放

    protected final boolean tryReleaseShared(int unused) {
        Thread current = Thread.currentThread();
    //如果当前线程是第一个获取读锁的线程（多次获取原因）
        if (firstReader == current) {
            // assert firstReaderHoldCount > 0;
            if (firstReaderHoldCount == 1)
                firstReader = null;
            else
                firstReaderHoldCount--;
        } else {    
        //如果当前线程不是第一个获取读锁的线程
        //(获取当前线程的HoldCounter）
            HoldCounter rh = cachedHoldCounter;
            if (rh == null || rh.tid != getThreadId(current))
                rh = readHolds.get();
            int count = rh.count;
            if (count <= 1) {
            //当前线程获取的读锁小于等于1那么就将remove当前线程的HoldCounter
                readHolds.remove();
             //当前线程获取的读锁小于等于0抛出异常   
                if (count <= 0)
                    throw unmatchedUnlockException();
            }
            --rh.count;
        }
        //自旋 
        for (;;) {
            int c = getState();
            int nextc = c - SHARED_UNIT;
            if (compareAndSetState(c, nextc))
                // Releasing the read lock has no effect on readers,
                // but it may allow waiting writers to proceed if
                // both read and write locks are now free.
                return nextc == 0;
        }
    }

读锁的释放

锁降级

锁降级指的是先获取到写锁，然后获取到读锁，然后释放了写锁的过程。
因为在获取读锁的时候的判断条件是：

if (exclusiveCount(c) != 0 &&
        getExclusiveOwnerThread() != current)
        return -1;

锁降级 锁降级必要

ReentrantReadWriteLock锁的特性：

• （1）读读共享；
• （2）写写互斥；
• （3）读写互斥；
• （4）写读互斥；

特性一 -> 同时获得锁

public class ReentrantReadWriteLockDemo {
    private ReentrantReadWriteLock lock = new ReentrantReadWriteLock();

    public static void main(String[] args) {

        ReentrantReadWriteLockDemo demo = new ReentrantReadWriteLockDemo();

        new Thread(() -> demo.read(), "ThreadA").start();
        new Thread(() -> demo.read(), "ThreadB").start();
    }

    private void read() {
        try {
            try {
                lock.readLock().lock();
                System.out.println("获得读锁" + Thread.currentThread().getName()
                        + " 时间:" + System.currentTimeMillis());
                //模拟读操作时间为5秒
                Thread.sleep(5000);
            } finally {
                lock.readLock().unlock();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
//    获得读锁ThreadA 时间:1513133068581
//    获得读锁ThreadB 时间:1513133068582


}

参考

1.死磕Java并发】—–J.U.C之重入锁：ReentrantLock；
2.方腾飞：《Java并发编程的艺术》
3.可重入读写锁ReentrantReadWriteLock基本原理分析
4.【Java并发】- ReentrantReadWriteLock，读写锁原理