《并发编程的艺术》阅读笔记之Lock与AQS

Lock接口

在jdk1.5之后，并发包下新增了一个lock接口，lock接口定义了锁的获取，锁的释放，等方法，需要用户手动设置。与关键字不同的是，lock具有可操作性，比如，可以中断线程，设置超时时间，如果时间截止，该线程还没有获得锁，就直接返回。同时lock接口有两个主要的实现类，可重入锁和读写锁。

java.util.concurrent.locks.Lock 接口，比 synchronized 提供更具拓展性的锁操作。它允许更灵活的结构，可以具有完全不同的性质，并且可以支持多个相关类的条件对象。

它的优势有(与synchronized区别)：

1. 可以使锁更公平。
2. 可以使线程在等待锁的时候响应中断。
3. 可以让线程尝试获取锁，并在无法获取锁的时候立即返回或者等待一段时间。
4. 可以在不同的范围，以不同的顺序获取和释放锁。

Lock接口的实现（如Reentrantlock）基本都是通过聚合了一个同步器的子类来完成线程访问控制的。

队列同步器AbstractQueuedSynchronizer

AQS是用来构建锁或者其他同步组件的基础框架，它使用了一个int成员变量标识同步状态，通过内置的FIFO队列来完成资源获取线程的排队工作。同步器面向的是锁的实现者，它简化了锁的实现方式，屏蔽了同步状态管理、线程的排队、等待与唤醒的底层操作。

使用方法

同步器的主要是用方式是继承，子类通过继承同步器并实现它的抽象方法来管理同步状态，在实现抽象方法时使用同步器提供的3个同步状态获取和释放方法（getstate,setstate,compareAndSetState）对同步状态进行更改，他们可以保证状态的改变是安全的。子类推荐被定义为自定义同步组件的静态内部类。同步器自身没有实现任何同步接口，它仅仅是定义了若干同步状态获取和释放的方法来供自定义同步组件使用。

用户通过使用AQS提供的getState、setState、compareAndSetState三个访问/修改同步状态的方法实现TryAcqurie、tryRelease、tryAcquiredShared、tryReleaseShared、isHeldExclusively等5个方法来实现一个AQS子类即可。

然后在实现自定义同步组件时只需要通过将操作代理到我们自己定义的AQS子类上，调用同步器提供的acquire等模板方法（模板方法会相应调用用户实现的方法），即可实现一个可靠的自定义同步组件。(自己总结)

原理

AQS是列队同步器。他是并发类的基础组件。它的内部维护了一个volitale修饰的state状态变量。当状态为0时，说明没有任何线程占有锁，当状态改为1时，说明有线程持有锁。其他线程获取同步状态失败会被封装为一个Node节点的数据结构，然后加入一个FIFO的双向同步队列尾部，加入后以自旋的方式尝试获取同步状态（判断前驱结点为头结点且获取同步状态成功），如获取失败会阻塞当前线程。

//同步器的Acquire方法
public final void acquire(int arg) {
    if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

//同步器的addWaiter和enq方法
private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode); // 快速尝试在尾部添加 Node pred = tail;
	if (pred != null) { 
        node.prev = pred; 
        if (compareAndSetTail(pred, node)) { 
            pred.next = node; return node; 
        } 
    }
    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; 
            } 
        } 
    } 
}

//同步器的acquireQueued方法
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)) { //唤醒后继续获取，前提是前驱结点必须是head
                setHead(node); 
                p.next = null; // help GC 
                failed = false; 
                return interrupted; 
            }if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())//获取失败则阻塞
                interrupted = true; 
        } 
    } finally { 
        if (failed) cancelAcquire(node); 
    } 
}

而被阻塞线程的唤醒主要依靠前驱节点的出队和阻塞线程的中断实现。当成功获取到同步状态的线程执行完业务逻辑，同步器调用tryRelease释放同步状态，之后唤醒（用的实际是LockSupport）头结点的后继结点中的线程。tryRelease不需要循环cas保证线程安全因为只会有一个线程持有同步状态。被唤醒的线程会把自己设置为新的头结点。

同步器拥有首节点和尾结点，线程被转换成节点加入队列（尾部）这个过程需要保证线程安全，因此同步器提供了一个基于CAS的设置尾结点的方法comPareandSetTail。

其中内部类还有一个conditionObject构建等待队列，将condition调用await方法时线程会加入到等待队列中，直到condition调用signal()方法后，线程会被唤醒从等待队列进入同步队列中进行锁的竞争。

以上所说为独占式获取和释放，共享式获取的逻辑和上面差不多，调用tryAcquiredShared方法返回值>=0则能够获取同步状态，获取失败则转换为节点进入等待队列尾部，加入后以自旋的方式尝试获取同步状态（判断前驱结点为头结点且获取同步状态成功），如获取失败会阻塞当前线程。

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

共享式中当有获取到同步状态的线程释放同步状态时，将会唤醒后续处于等待状态的节点。这块跟独占式的区别主要在于tryReleaseShared方法必须通过循环CAS确保同步状态线程安全释放，因为释放同步状态的操作会来自多个线程。

ReentrantLock

ReentrantLock，可重入锁，是一种递归无阻塞的同步机制。它可以等同于 synchronized 的使用，但是 ReentrantLock 提供了比 synchronized 更强大、灵活的锁机制，可以减少死锁发生的概率。

ReentrantLock有公平锁和非公平锁两种实现方式。

非公平锁

//ReentrantLock的nonfairTryAcquire方法
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) throw new Error("Maximum lock count exceeded"); 
        setState(nextc); 
        return true; 
    }
    return false; 
}

CAS获取同步状态是常规操作，为了让当前持有锁的线程再次获取锁不被阻塞，ReentrantLock在实现nonfairTryAcquire时判断线程是否持有锁，持有的话就增加state值，并return true。

//ReentrantLock的tryRelease方法
protected final boolean tryRelease(int releases) { 
    int c = getState() - releases; //因为只有一个线程持有锁，所以不需要用CAS
    if (Thread.currentThread() != getExclusiveOwnerThread()) 
        throw new IllegalMonitorStateException(); 
    boolean free = false; 
    if (c == 0) { 
        free = true; 
        setExclusiveOwnerThread(null); 
    }
    setState(c); 
    return free; 
}

释放的时候判断是否state变为了0，只有为0才返回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; 
}

唯一得区别是公平锁调用 #hasQueuedPredecessors() 方法检查是否有前序节点，有则返回false。

非公平锁和公平锁获取同步状态的过程，会发现两者唯一的区别就在于，公平锁在获取同步状态时多了一个限制条件 <1> 处的 #hasQueuedPredecessors() 方法，是否有前序节点，即自己不是首个等待获取同步状态的节点。如果是则返回 true ，否则返回 false 。

读写锁

java并发包提供的读写锁实现为ReentrantReamWriteLock

读写状态设计

读写锁同样依赖自定义同步器来实现同步功能，而读写状态就是其同步器的同步状态。回想ReentrantLock中自定义同步器的实现，同步状态表示锁被一个线程重复获取的次数，而读 写锁的自定义同步器需要在同步状态（一个整型变量）上维护多个读线程和一个写线程的状 态，使得该状态的设计成为读写锁实现的关键。

如果在一个整型变量上维护多种状态，就一定需要“按位切割使用”这个变量，读写锁将 变量切分成了两个部分，高16位表示读，低16位表示写，划分方式如下图所示。

当前同步状态表示一个线程已经获取了写锁，且重进入了两次，同时也连续获取了两次 读锁。

读写锁是如何迅速确定读和写各自的状态呢？

答案是通过位运算。假设当前同步状态 值为S

写状态等于S&0x0000FFFF（将高16位全部抹去），

读状态等于S>>>16（无符号补0右移 16位）。当写状态增加1时，等于S+1，当读状态增加1时，等于S+(1<<16)，也就是 S+0x00010000。

根据状态的划分能得出一个推论：S不等于0时，当写状态（S&0x0000FFFF）等于0时，则读 状态（S>>>16）大于0，即读锁已被获取。这一点在源码中用于获取写锁时判断读锁。

写锁的获取与释放

//ReentrantReadWriteLock的tryAcquire方法
protected final boolean tryAcquire(int acquires) { 
    Thread current = Thread.currentThread(); 
    int c = getState(); //状态值
    int w = exclusiveCount(c); 
    if (c != 0) { // 存在读锁或者当前获取线程不是已经获取写锁的线程，即上文说的推论
        if (w == 0 || current != getExclusiveOwnerThread()) 
            return false; 
        if (w + exclusiveCount(acquires) > MAX_COUNT) 
            throw new Error("Maximum lock count exceeded"); 
        setState(c + acquires); 
        return true; 
    }
    if (writerShouldBlock() || !compareAndSetState(c, c + acquires)) { 
        return false; 
    }
    setExclusiveOwnerThread(current); 
    return true; 
}

写锁的获取：

写锁是一个支持重进入的排它锁。如果当前线程已经获取了写锁，则增加写状态。如果当前线程在获取写锁时，读锁已经被获取（读状态不为0）或者该线程不是已经获取写锁的线程， 则当前线程进入等待状态。

写锁的释放：

写锁的释放与ReentrantLock的释放过程基本类似，每次释放均减少写状态即可。当写状态为0 时表示写锁已被释放，从而等待的读写线程能够继续访问读写锁，同时前次写线程的修改对后续读写线程可见。

读锁的获取与释放

读锁是一个支持重进入的共享锁，它能够被多个线程同时获取。

//ReentrantReadWriteLock的tryAcquireShared方法
protected final int tryAcquireShared(int unused) { 
    for (;;) {
        int c = getState(); 
        int nextc = c + (1 << 16); 
        if (nextc < c) 
            throw new Error("Maximum lock count exceeded"); 
        if (exclusiveCount(c) != 0 && owner != Thread.currentThread()) 
            return -1; 
        if (compareAndSetState(c, nextc)) 
            return 1; 
    } 
}

读锁的获取：

在tryAcquireShared(int unused)方法中，如果其他线程已经获取了写锁，则当前线程获取读锁失败，进入等待状态。如果当前线程获取了写锁或者写锁未被获取，则当前线程（线程安全，依靠CAS保证）增加读状态，成功获取读锁。（总结，只要其他线程没有获取到写锁，并且读锁数量没超出最大值就行）

读锁的释放：

读锁的每次释放（线程安全的，可能有多个读线程同时释放读锁）均减少读状态，减少的 值是（1<<16）。

锁降级

锁降级指的是写锁降级成为读锁。但是锁降级要求在先不释放写锁的前提下获取读锁，然后再释放写锁。这样做的目的在于保证当前线程在降级到读锁之前不会有其他线程修改数据（如果先释放写锁再获取读锁，在当前线程释放写锁之后获取到读锁之前有可能其他线程获取到写锁修改了数据，那样当前线程将无法感知到数据修改，从而造成问题）。

备注：RentrantReadWriteLock不支持锁升级（把持读锁、获取写锁，最后释放读锁的过程）。目的 也是保证数据可见性，如果读锁已被多个线程获取，其中任意线程成功获取了写锁并更新了 数据，则其更新对其他获取到读锁的线程是不可见的（这里的不可见我的理解是不可感知）。评注：我觉得这个锁升级完全就不合理呀，好几个线程都拿着读锁，按照上面写锁获取tryAcquire的逻辑存在读锁也不能获取到写锁啊。

Condition

用途

Condition接口也提供了Object类似的监视器方法，与Lock配合可以实现等待/通知模式，Condition方法使用的前置条件为获取锁。

与synchronized的等待唤醒机制相比Condition具有更多的灵活性以及精确性，这是因为notify()在唤醒线程时是随机(同一个锁)，而Condition则可通过多个Condition实例对象建立更加精细的线程控制，也就带来了更多灵活性了，我们可以简单理解为以下两点。

  通过Condition能够精细的控制多线程的休眠与唤醒。

  对于一个锁，我们可以为多个线程间建立不同的Condition。线程之间的通信(等待唤醒机制，类似wait/notify)。

Object监视器模型上，一个对象拥有一个同步队列和等待队列，而并发包中的Lock（更确切地说是同步器）拥有一个同步队列和多个等待队列。

实现

ConstionObject是AQS的内部类，每个condition对象都包含一个队列（等待队列），Condition拥有首节点（firstWaiter）和尾节点 （lastWaiter）。Condition定义了等待/通知两种类型的方法，当前线程调用这些方法时，需要提前获取到 Condition对象关联的锁。一个Condition包含一个等待队列。

public final void await() throws InterruptedException { 
    if (Thread.interrupted()) 
        throw new InterruptedException(); 
    Node node = addConditionWaiter(); // 当前线程加入等待队列 
    int savedState = fullyRelease(node); // 释放同步状态，也就是释放锁 
    int interruptMode = 0; 
    while (!isOnSyncQueue(node)) { //await调用后和signal调用后从这被唤醒
        LockSupport.park(this); 
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) break; 
    }
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE) 
        interruptMode = REINTERRUPT; 
    if (node.nextWaiter != null) 
        unlinkCancelledWaiters(); 
    if (interruptMode != 0) 
        reportInterruptAfterWait(interruptMode); 
}

Await方法其实做了三件事

1. 是将当前线程封装node节点加入到等待队列中，
2. 是释放同步状态并且唤醒（同步队列的）后继节点的线程，
3. 是循环判断当前线程是否在同步队列中。如果没有就挂起当前线程，从而当前线程进入等待状态。（这步说的有点乱，其实是加入等待队列后这个判读条件肯定会false，然后就阻塞当前线程进入等待状态）因为等待线程被唤醒时，会从等待队列加入到同步队列中。

public final void signal() { 
    if (!isHeldExclusively()) 
        throw new IllegalMonitorStateException(); 
    Node first = firstWaiter; 
    if (first != null) 
        doSignal(first); //只需要持有锁的线程将等待队列中的首节点移入同步队里然后唤醒，之后被唤醒几点会从上文的await方法中换新，执行acquireQueued在其中循环竞争，代码非常巧妙
}

Signal方法主要做了两件事

1. 是判断当前线程是否持有锁（独占锁），如果没有就抛出异常。因为独占模式才会放入等待队列中。

2. 获取等待队列中的首节点，将其移动到同步队列并使用LockSupport唤醒该节点中的线程

被唤醒的线程将从await()方法中的while循环中退出（即上面说到的await第三件事），进而调用同步器的acquireQueued()方法加入到获取同步状态的竞争中。

condition的signalAll()方法相当于对等待队列中的每个节点均执行一次signal()方法，效果就是将等待队列中的所有节点全部移动到同步队列中，并唤醒每个节点的线程。