《java并发编程实战笔记》

• 第十五章 原子变量与非阻塞同步机制
• 现有的锁的劣势
• 比较交换指令CAS
• 非阻塞计数器(此部分有误)
• 原子变量是一种"更好的volatile"
• 锁与原子变量的性能比较
• 非阻塞算法(Nonblocking Algorithms)
• 非阻塞的栈
• 非阻塞的链表
• 原子的域更新器
• ＡＢＡ问题

第十五章 原子变量与非阻塞同步机制

  近年来，在并发算法领域的大多数研究都侧重于非阻塞算法，这种算法采用底层的原子机器指令，例如比较交换指令来代替锁确保数据在并发访问中的一致性。
  非阻塞算法广泛运行在操作系统和JVM实现进程、垃圾回收机制以及锁和其他并发数据结构，非阻塞算法可以使多个线程在竞争相同的数据时不会发生阻塞，并且极大的减少调度开销。

现有的锁的劣势

一：无论哪个线程持有守护变量的锁，都只能采用独占方式来访问这些变量。
二：如果有多个线程同时请求锁，那么一些线程将被挂起或者采取自旋方式并在在稍后恢复运行，然而在挂起和恢复过程中存在很大的开销。(虽然说采用volatile变量不会发生上下文切换或线程调度等操作，并且也能保证可见性，但是volatile不能用于构建原子的复合操作)
三：当一个线程正在等待锁时，它不能做其他任何事情。 如果一个线程在持有锁的情况下被延迟执行（例如发生了缺页错误，调度延迟，或者其他类似情况），那么所有需要这个锁的线程都无法执行下去。
  如果被阻塞线程的优先级较高，而持有锁的线程优先级较低，那么这是严重的，被称为优先级反转（Priority Inversion）。即使高优先级的线程可以抢先执行，但仍需要等待锁被释放，从而导致它的优先级会降至低优先级的级别。如果持有锁的线程被永久阻塞，所有等待这个锁的线程就永远无法执行下去。

比较交换指令CAS

  大多数的处理器采用的原子机器指令实现非阻塞算法是用比较交换指令CAS。CAS包含了3个操作数——需要读写的内存位置V，进行比较的值A和拟写入的新值B。当且仅当V的值等于A时，CAS才会通过原子方式用新值B来更新V的值，否则不会执行任何操作。 无论位置V的值是否等于A，都将返回V原有的值（这种变化形式被称为比较并设置，无论操作是否成功都会返回）。CAS的含义是：“我认为V的值应该是A，如果是，那么将V的值更新为B，否则不修改并告诉V的实际为多少”。SimulatedCAS 是模拟CAS操作，真正的CAS实现是通过原子机器指令。
  当多个线程尝试使用CAS同时更新同一个变量时，只有其中一个线程能更新变量的值，而其他线程都将失败。然而，失败的线程并不会被挂起（这与获取锁不同：当获取锁失败时，线程将被挂起），而是被告知在这次竞争中失败，并可以再次尝试。
  由于一个线程在竞争CAS时失败不会阻塞，因此它可以决定是否重新尝试，或者执行一些恢复操作，或者不执行任何操作（在非阻塞算法中，当CAS失败时，意味着其他线程已经完成了你想执行的操作）。

@ThreadSafe
public class SimulatedCAS {
private int value;
public synchronized  int get(){
return value;
}
public synchronized int compareAndSwap(int expectedValue,int newValue){
int oldValue=value;
if(oldValue==expectedValue)
value=newValue;
return oldValue;
}
public synchronized boolean compareAndSet(int expectedValue,int newValue){
//如果相同就将值设置为newValue。并返回true
return (expectedValue==compareAndSwap(expectedValue, newValue));
}
}

非阻塞计数器(此部分有误)

  个人认为本书此一小部分有误，书中原话“CasCounter不会阻塞，但如果其他线程同时更新计数器，那么会多次执行重试操作”。CasCounter例子采用的是SimulatedCAS 并不是真正的CAS实现的，SimulatedCAS 偌大的synchronized 方法在执行时需要获得SimulatedCAS 对象的锁，还说CasCounter是非阻塞式计数器。 CAS部分暂时先不进一步学习。

@ThreadSafe
public class CasCounter {
private SimulatedCAS value;
public int getValue(){
return value.get();
}
public int increment(){
int v;
do{
v=value.get();
}
//只要不符合条件，即过程中没有被其他线程抢先操作就一直循环
while(v!=value.compareAndSwap(v, v+1));
return v+1;
}
}

原子变量是一种"更好的volatile"

  原子变量比锁的粒度更细，量级更轻，并且对于在多处理器系统上实现高性能的并发代码来说是非常关键的。原子变量将发生竞争的范围缩小到单个变量上，这是你获得的粒度最小的情况。 更新原子变量的快速（非竞争）路径比获取锁的快速路径块，而慢速路径也一样，因为它不需要挂起或重新调度线程。 在使用基于原子变量而非锁的算法中，线程在执行时更不易出现延迟，并且如果遇到竞争，也更容易恢复过来。
  原子变量相当于一种泛化的volatile变量，能够支持原子的和有条件的读-该-写操作。原子变量共有12个原子变量类，可分为4组：标量类（scalars），更新器类（field updaters），数组类以及复合变量类（compound variables）。 其中最常用的原子变量就是标量类：AtomicInteger，AtomicLong，AtomicBoolean以及AtomicReference。所有原子变量类都支持CAS，此外，AtomicInteger和AtomicLong还支持算法运算。
   AtomicInteger表示一个int类型的值，并提供了get和set方法，这些Volatile类型的int变量在读取和写入上有着相同的内存语义。 它还提供了一个原子的compareAndSet方法（如果该方法成功执行，那么将实现与读取、写入一个volatile变量相同的内存效果），以及原子的添加，递增递减等方法。
  原子数组类（只支持Integer，Long和Reference）中的元素可以实现原子更新。原子数组类为数组的元素提供了volatile类型的访问语义。基本变量类是不可修改的，而原子变量时可修改的。
  我们曾经用不可变对象volatile引用来原子地更新多个状态变量，此处我们可以通过原子变量来维护多个变量的不变性条件。

//   15-3   通过CAS来维持包含多个变量的不变性条件
@ThreadSafe
public class CasNumberRange {
private static class IntPair{
final int lower;   //不变性条件，lower<upper
final int upper;
public IntPair(int lower,int upper){
this.lower=lower;
this.upper=upper;
}
}
private final AtomicReference<IntPair> values=
new AtomicReference<IntPair>(new IntPair(0, 0));
public int getLower(){
return values.get().lower; //get得到引用，IntPair类型
}
public int getUpper(){
return values.get().upper;
}
public void setLower(int i){
while(true){
IntPair oldv=values.get();
if(i>oldv.upper)
throw new IllegalArgumentException("Can't set lower to " + i + " > upper");
IntPair newv=new IntPair(i, oldv.upper);
if(values.compareAndSet(oldv, newv))
return;
}
}
public void setUpper(int i) {
while (true) {
IntPair oldv = values.get();
if (i < oldv.lower)
throw new IllegalArgumentException("Can't set upper to " + i + " < lower");
IntPair newv = new IntPair(oldv.lower, i);
if (values.compareAndSet(oldv, newv))
return;
}
}
}

锁与原子变量的性能比较

  为了说明锁与原子变量之间的可伸缩性差异，我们将比较伪随机数字生成器(PRNG，pseudorandom number generator)的几种不同实现。在PRNG中，在生成下个随机数字时需要用到上一个数组，所以在PRNG中必须记录前一个数值并将其作为状态的一部分。15-4和15-5给出了线程安全的PRNG的两种实现，一种基于ReentrantLock，一种基于AtomicInteger。 测试程序将反复调用它们，在每次迭代中将生成一个随机数字（在此过程中将读取并修改共享的seed状态），并执行一些仅在线程本地数据上执行的“繁忙”迭代。

public class PseudoRandom {
int calculateNext(int prev) {
prev ^= prev << 6;
prev ^= prev >>> 21;
prev ^= (prev << 7);
return prev;
}
}

基于ReentrantLock实现的随机数生成器：

//    15-4  基于ReentrantLock实现的随机数生成器
@ThreadSafe
public class ReentrantLockPseudoRandom extends PseudoRandom {
private final Lock lock = new ReentrantLock(false);
private int seed;
ReentrantLockPseudoRandom(int seed) {
this.seed = seed;
}
public int nextInt(int n) {
lock.lock();
try {
int s = seed;
seed = calculateNext(s);
int remainder = s % n;
return remainder > 0 ? remainder : remainder + n;
} finally {
lock.unlock();
}
}
}

基于AtomicInteger实现的RseudoRandom：

//   15-5  基于AtomicInteger实现的RseudoRandom
@ThreadSafe
public class AtomicPseudoRandom extends PseudoRandom{
private AtomicInteger seed;
AtomicPseudoRandom(int seed) {
this.seed=new AtomicInteger(seed);
}
public int nextInt(int n){
while(true){
int s=seed.get();
int nextSeed=calculateNext(s);
if(seed.compareAndSet(s, nextSeed)){
int remainder=s%n;
return remainder>0?remainder:remainder+1;
}
}
}
}

  下面给出在每次迭代中工作量较低以及适中情况下的吞吐量。 如果线程本地的计算量较少，那么在锁和原子变量上的竞争将非常激烈；如果线程本地的计算量较多，那么在锁和原子变量上的竞争会降低，因为在线程中访问锁和原子变量的频率将减低。
                  
  可以看出，在高度竞争的情况下，锁的性能将超过原子变量的性能，但在更真实的情况下，原子变量的性能将超过锁的性能。 这是因为锁在发生竞争时会挂起线程，从而降低了CPU的使用率和共享内存总线上的同步信号量。（类似于生产者-消费者模式中的可阻塞生产者，它能降低消费者上的工作负载，使消费者的处理速度赶上生产者的处理速度）。
  另一方面，使用原子变量，那么发出调用的类负责对竞争进行管理，与大多数基于CAS的算法一样，AtomicPseudoRandom遇到竞争时将立即重试，在激烈竞争环境会导致了更多的竞争。
  在实际情况中，原子变量在可伸缩性上要高于锁，因此在应对常见的竞争程度时，原子变量的效率会更高。在中低程度的竞争下，原子变量能提供更高的可伸缩性，而在高强度的竞争下，锁能更有效地避免竞争。
  图中还包含第三条曲线，它时一个使用ThreadLocal来保存PRNG状态的PseudoRandom，这种实现方法改变了类的行为，即每个线程都只能看到自己私有的伪随机数字序列，而不是所有线程共享一个随机数序列。 这说明了，如果能避免使用共享状态，那么开销将更小。 我们可以通过提高处理竞争的效率来提高可伸缩性，但只有完全消除竞争，才能实现真正的可伸缩性。

非阻塞算法(Nonblocking Algorithms)

如果在某种算法中，一个线程的失败或挂起不会导致其他线程也失败或挂起，这种算法被称为非阻塞算法。 在实现相同功能的前提下，非阻塞算法通常比基于锁的算法更为复杂。

非阻塞的栈

  栈是最简单的链式数据结构，每个元素仅指向一个元素，并且每个元素也只被一个元素引用。15-6给出了如果通过原子引用来构建栈的示例。 栈是由Node元素构成的一个链表，其中栈顶作为根节点，并且在每个元素中都包含了一个值以及指向下一个元素的链接。push方法创建一个新的节点，该节点的next域指向当前的栈顶，然后使用CAS把这个新节点放入栈顶。如果在开始插入节点时，位于栈顶的节点没有发生变化，那么CAS就会成功。如果栈顶节点发生了变化（例如由于其他线程在本线程开始之前插入或移除了元素），那么CAS将会失败，而push方法会根据栈的当前状态来更新节点，并且再次尝试。无论哪种情况，在CAS执行完成后，后栈仍会处于一致的状态。

//   15-6  使用Treiber算法构造的非阻塞栈
@ThreadSafe
public class ConcurrentStack<E>{
AtomicReference<Node<E>> top=new AtomicReference<Node<E>>();
//push方法创建一个新的节点，该节点的next域指向当前的栈顶，然后使用CAS把这个新节点放入栈顶。
//如果在开始插入节点时，位于栈顶的节点没有发生变化，那么CAS就会成功。
public void push(E item){  //根据栈的当前状态来更新节点
Node<E> newHead=new Node<E>(item);
Node<E> oldHead;
do{
oldHead=top.get();
newHead.next=oldHead;
}while(!top.compareAndSet(oldHead, newHead));//如果栈顶点发生变化则CAS失败,重新执行
}
public E pop(){
Node<E> oldHead;
Node<E> newHead;
do{
oldHead=top.get();
if(oldHead==null)
return null;
newHead=oldHead.next;
}while(!top.compareAndSet(oldHead, newHead)); //如果栈顶点发生变化则CAS失败,重新执行
return oldHead.item;
}
private static class Node <E> {
public final E item;
public Node<E> next;  //下一个节点
public Node(E item) {
this.item = item;
}
}
}

  CasCounter和ConcurrentStack说明了非阻塞算法的所有特性: 某项工作的完成具有不确定性，必须重新执行。像ConcurrentStack这样的阻塞算法中都能确保线程安全性，因为CAS像锁定机制一样，既能提供原子性，又能提供可见性。
  当一个线程需要改变栈的状态时，将调用CAS，这个方法与写入volatile类型的变量有着相同的内存效果。
  当线程检查栈的状态时，将在同一个AtomicReference上调用get方法，这个方法与读取volatile类型的变量有着相同的内存效果。
  因此，一个线程的任何修改结构都可以安全地发布给其他正在查看状态的线程。并且，这个栈时通过CAS来修改的，因此将采用原子操作来更新top的引用，或者在发现存在其他线程干扰的情况下，修改操作将失败。

非阻塞的链表

  CAS的基本使用模式：在更新某个值时存在不确定性时，以及在更新失败时重新尝试。链接队列比栈复杂，因为它必须支持对头结点和尾结点的快速访问。因此，它需要单独维护的头指针和尾指针。 有两个指针指向位于尾部的节点：当前最后一个元素的next指针，以及尾结点。当成功地插入一个新元素时，这两个指针都需要使用原子操作来更新。

  在更新这两个指针时需要不同的CAS操作，并且如果第一个CAS成功，但第二个失败，那么队列将处于不一致的状态。而且，即使两个CAS都成功了，在执行两者之间仍有可能有另一个线程会访问这个队列。
  我们需要一些技巧。第一个技巧是，即使在一个包含多个步骤的更新操作中，也要确保数据结构总是处于一致的状态。这样，当线程B到达时，如果发现线程A正在执行更新，B不能立即开始执行自己的更新操作，等待（通过反复检查队列的状态）并直到A完成更新，从而使两个线程不会互相干扰。这有可能有一个线程更新操作失败了，其他的线程都无法访问队列。要使该算法成为一个非阻塞的方法，必须确保一个线程失败时不会妨碍其他线程继续执行下去。
  因此，第二个技巧是，如果当B到达时发现A正在修改数据结构，那么在数据结构中应该有足够多的信息，使得B能完成A的更新操作。如果B”帮助”A完成了更新操作，那么B可以执行自己的操作，而不用等待A的操作完成。当A恢复后在试图完成其操作时，发现B已经替它完成了。15-7给出了非阻塞链接队列算法中的插入部分。在许多队列算法中，空队列通常包含一个“哨兵（Sentinel）”节点或“哑（Dummy）节点”，并且头节点和尾节点在初始化时都指向该哨兵节点。尾节点通常要么指向哨兵节点（如果队列为空），即队列的最后一个元素，要么（当有操作正在执行更新时）指向倒数第二个元素。
  下图给出一个处于正常（稳定）状态的包含两个元素的队列 ：

//   15-7  非阻塞算法中的插入算法
@ThreadSafe
public class LinkedQueue <E>{
private static class Node<E>{
final E item;
final AtomicReference<Node<E>> next;

public Node(E item,Node<E> next){
this.item=item;
this.next=new AtomicReference<LinkedQueue.Node<E>>(next);
}
}
private final Node<E> dummy=new Node<E>(null,null);  //哑结点
private final AtomicReference<Node<E>> head=
new AtomicReference<Node<E>>(dummy);          //头结点
private final AtomicReference<Node<E>> tail=
new AtomicReference<Node<E>>(dummy);          //尾节点
public boolean put(E item){
Node<E> newNode=new Node<E>(item,null);
while(true){
Node<E> curTail=tail.get();
Node<E> tailNext=curTail.next.get();
if(curTail==tail.get()){
if(tailNext!=null){   //A
//队列处于中间状态（即插入成功但未推进尾节点），推进尾节点
tail.compareAndSet(curTail, tailNext);   //B
}else{
//处于稳定状态(tailNext==null)，尝试插入新节点
//如果两个线程同时插入元素，那么这个CAS将失败。在这个情况下，并不会造成破坏，不会发生任何变化，并且当前的线程只需重新读取尾节点并再次重试。
if(curTail.next.compareAndSet(null, newNode)){       //C
//插入成功，尝试推进尾节点
tail.compareAndSet(curTail, newNode);       //D
//如果步骤D失败，那么执行插入操作的线程将返回，而不是重新执行CAS，因为另一个线程在步骤B中完成了这个工作
return true;
}
}
}
}
}
}

                  
  当插入一个新元素时，需要更新两个指针。首先更新当前最后一个元素的next指针，将新节点链接到队列末尾，然后更新尾节点，将其指向这个新元素。 在这两个操作之间，队列处于一个中间状态(此时tailNext!=null)， 如图:
                  
  在第二次更新完成后，队列将再次处于稳定状态(尾节点的next域为空)。在插入新元素之前，首先检查队列是否处于中间状态（步骤A），如果是，那么有另一个线程正在插入元素(步骤C和D之间)。此时当前线程不会等待其他线程执行完成，而是帮助它完成操作，并将尾节点向前推进一个节点(步骤B)，然后它将重复执行这种检查，以免另一个线程已经开始插入新元素，并继续推进尾节点，直到它发现队列处于稳定状态之后，才会执行自己的插入操作.
  步骤C中的CAS把新节点链接到队列尾部，因此如果两个线程同时插入元素，那么这个CAS将失败。 在这个情况下，并不会造成破坏，不会发生任何变化，并且当前的线程只需重新读取尾节点并再次重试。
  如果步骤C成功，那么插入操作生效，第二个CAS(步骤D)被认为是一个”清理操作”，既可以由执行插入操作的线程来执行，也可以由其他任何线程来执行。
  如果步骤D失败，那么执行插入操作的线程将返回，而不是重新执行CAS，因为另一个线程在步骤B中完成了这个工作。
  这种方式能够工作，因为在任何线程尝试将一个新节点插入到队列之前，都会首先通过检查tail.next是否非空来判断是否需要清理队列。如果是，它首先会推进尾节点（可能需要执行多次），直到队列处于稳定状态。

原子的域更新器

  ConcurrentLinkedQueue没有使用原子引用来表示每个Node，而是使用普通的volatile类型引用，并通过基于反射的AtomicReferenceFieldUpdater来进行更新，如15-8。原子的域更新器类AtomicReferenceFieldUpdater表示现有volatile域的一种基于反射的“视图”，从而能够在已有的volatile域上使用CAS。在更新器类中没有构造函数，要创建一个更新器对象，可以调用newUpdater方法，并制定类与域的名字。域更新器类没有与某个特定的实例关联在一起，因而可以更新目标类的任意实例中的域。更新器类提供的原子性保证比普通原子类更弱一些，因为无法保证底层的域不被直接修改——CAS以及其他算法方法只能确保其他使用原子域更新器方法的线程的原子性。

//       在ConcurrentLinkedQueue中使用原子的域更新器
private class Node<E>{
private final E item;
private volatile Node<E> next;
public Node(E item){
this.item=item;
}
}
//在更新器类中没有构造函数，要创建一个更新器对象，可以调用newUpdater方法
private static AtomicReferenceFieldUpdater<Node,Node> nextUpdater=AtomicReferenceFieldUpdater.newUpdater(Node.class,Node.class,"next");

ＡＢＡ问题

  ABA问题:如果在算法中的节点可以被循环使用，那么在使用”比较并交换”指令时就可能出现这种问题(主要在没有垃圾回收机制的环境中)。在CAS操作中将判断“V的值是否仍为A”，并且如果是的话就继续执行更新操作。有时候还需直到“自从上次看到V的值为A以来，这个值是否发生了变化”。在某些算法中，如果V的值首先由A变成B，再由B变成A，那么仍然被认为是发生了变化，并需要重新执行算法中的某些步骤。
  如果在算法中采用自己的方式来管理节点对象的内存，那么可能出现ABA问题。在这种情况下，即使链表的头节点仍然指向之前观察到的节点，那么也不足与说明链表的内容没有发生变化。
  如果通过垃圾回收器来管理链表节点仍然无法避免ABA问题，那么还有一个相对简单的方案：不是更新某个引用的值，而是更新两个值，包括一个引用和一个版本号。即使这个值由A变B再变A，版本号也是不同的。AtomicStampedReference(以及AtomicMarkableReference)支持这两个变量上执行原子的条件更新。 AtomicStampedReference将更新一个”对象-引用”二元组，通过在引用上加上“版本号”，避免ABA问题。 AtomicMarkableReference将更新一个“对象引用-布尔值”二元组。

小结

  非阻塞算法通过底层的并发原语(例如比较交换而不是锁)来维持线程安全性。这些底层的原语通过原子变量类向外公开，这种类也用做一种”更好的volatile变量”，从而为整数和对象引用提供原子的更新操作。