Java并发编程实战第15章原子变量和非阻塞同步机制

非阻塞的同步机制

简单的说，那就是又要实现同步，又不使用锁。

与基于锁的方案相比，非阻塞算法的实现要麻烦的多，但是它的可伸缩性和活跃性上拥有巨大的优势。

实现非阻塞算法的常见方法就是使用volatile语义和原子变量。

硬件对并发的支持

原子变量的产生主要是处理器的支持，最重要的是大多数处理器架构都支持的CAS（比较并交换）指令。

模拟实现AtomicInteger的++操作

首先我们模拟处理器的CAS语法，之所以说模拟，是因为CAS在处理器中是原子操作直接支持的。不需要加锁。

public synchronized int compareAndSwap(int exceptValue, int newValue){
int oldValue = count;
if(oldValue == exceptValue){
count = newValue;
}
return oldValue;
}

注意：CAS总是返回oldValue。

使用上面的方法模拟AtomicInteger的++操作。

class MyAtomicInteger{
private int value;
public int incrementAndGet()
{
int v ;
do{
v = value;
}while(v != compareAndSwap(v,v+1));
return v + 1;
}
}

注意：Java的AtomicInteger大概实现机制就是这样的，不会阻塞，使用处理器的CAS功能，但是要轮询尝试。

看起来轮询尝试性能会更差，其实不然，当竞争不是非常高的时候，基于CAS的算法更胜一筹。

原子类

AtomicBoolean AtomicInteger AtomicLong AtomicReference

原子数组类：AtomicIntegerArray AtomicLong AtomicReferenceArray。

使用volatile语法修饰的数组只能保证数组变量本身的volatile语义，不能保证元素的volatile语义。这个时候应该使用，原子数组类。

注意：AtomicBoolean AtomicInteger AtomicLong和非原子的对应数值类如Integer截然不用。实现机制完全不一样，也没有对应关系。最重要的一个差别：这个三个原子类是可变的。而且是使用的Object的hashCode和equals方法，没有自己扩展。

性能比较：锁与原子变量

在中低程度的竞争下，原子变量能提供很高的可伸缩性，原子变量性能超过锁；而在高强度的竞争下，锁能够更有效地避免竞争，锁的性能将超过原子变量的性能。但在更真实的实际情况（一般没有那么高强大的竞争）中，原子变量的性能将超过锁的性能。

注意：不论是锁还是原子变量，都远远比不上避免共享状态（如使用线程封闭技术，但是使用场景计较局限），彻底消除竞争的效率。

两个非阻塞的算法示例

/**
* 使用Treiber算法构造的非阻塞栈
*/
public class ConcurrentStack<E> {
private AtomicReference<Node<E>> top = new AtomicReference<ConcurrentStack.Node<E>>();
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));
}
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));
return oldHead.item;
}
private static class Node<E>{
public final E item;
public Node<E> next;
public Node(E item){
this.item = item;
}
}
}

下面这个没有看懂，留在这里记录，以后再看：

/**
* 链表中非阻塞算法中的插入排序，来自Michael-Scott
*/
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<>(next);
}
}
private final Node<E> dummy = new Node<E>(null, null);
private final AtomicReference<Node<E>> head =
new AtomicReference<>(dummy);
private final AtomicReference<Node<E>> tail =
new AtomicReference<>(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){
// 队列处于中间状态(即新节点已经接上，尾节点还未更新)，推进尾节点
tail.compareAndSet(curTail, tailNext);
} else{
// 处于稳定状态, 尝试插入新节点
if (curTail.next.compareAndSet(null, newNode)){
// 插入成功后，推进尾节点
tail.compareAndSet(curTail, tailNext);
return true;
}
}
}
}
}
}