并发编程 21—— 原子变量和非阻塞同步机制

Java并发编程实践 目录

并发编程 01—— ThreadLocal

并发编程 02—— ConcurrentHashMap

并发编程 03—— 阻塞队列和生产者-消费者模式

并发编程 04—— 闭锁CountDownLatch 与 栅栏CyclicBarrier

并发编程 05—— Callable和Future

并发编程 06—— CompletionService : Executor 和 BlockingQueue

并发编程 07—— 任务取消

并发编程 08—— 任务取消 之 中断

并发编程 09—— 任务取消 之 停止基于线程的服务

并发编程 10—— 任务取消 之 关闭 ExecutorService

并发编程 11—— 任务取消 之 “毒丸”对象

并发编程 12—— 任务取消与关闭 之 shutdownNow 的局限性

并发编程 13—— 线程池的使用 之 配置ThreadPoolExecutor 和 饱和策略

并发编程 14—— 线程池 之 整体架构

并发编程 15—— 线程池 之 原理一

并发编程 16—— 线程池 之 原理二

并发编程 17—— Lock

并发编程 18—— 使用内置条件队列实现简单的有界缓存

并发编程 19—— 显式的Conditon 对象

并发编程 20—— AbstractQueuedSynchronizer 深入分析

并发编程 21—— 原子变量和非阻塞同步机制

概述

第1部分 锁的劣势

第2部分 硬件对并发的支持

第3部分 原子变量类

第4部分 非阻塞算法

4.1 非阻塞的栈

4.2 非阻塞的链表

　4.3 ABA问题

参考

第1部分 锁的劣势

　　java传统上是使用锁来实现线程间的同步。每一个object都有一个内部锁(intrinsic lock)，关键字synchronized实现了锁的互斥。方法Object.wait主动的放弃锁，而方法Object.notify(All)则用来唤醒waiting中的线程。使用synchronized确保线程同步的正确性，但是也容易造成线程的阻塞。

　　volatile变量与锁相比是更轻量级的同步机制，但仅仅能保证内存的可见性，而不能用于原子化的操作。++i 看起来是原子的，而实际上却是取当前值，自增加一，然后回写更新。

　　使用锁有以下的缺点：

(1) 如果线程持有锁而延迟，会导致其他的线程的等待。

(2) 高优先级的线程阻塞，而低优先级的线程持有锁造成 优先级反转(priority inversion)。

(3) 如果持有锁的线程被永久地阻塞，所有等待这个锁的线程就无法执行下去，造成 死锁(dead lock)。

第2部分 硬件对并发的支持

　　独占锁是一种悲观的技术，总是假设闯入的线程会改变共享的变量。而对于细粒度的操作应该采用的乐观的解决方法。容许其他线程的闯入，但是在需要改变共享变量之前总是要检测其是否被修改。如果没有被修改，则完成更新，否则就放弃更新。CAS(compare and swap)就是这样的一种技术，现在已经被大多数处理器直接支持。

　　CAS的含义是：“我认为V的值应该是A，如果是，那么将V的值更新为B，否则不修改并告诉V的值实际为多少”。CAS是一项乐观的技术，它希望能成功地执行更新操作，并且如果有另外一个线程在最近一次检查后更新了该变量，那么CAS能检测到这个错误。

CAS的模拟：

@ ThreadSafe
public class SimulatedCAS {
@ GuardeBy( "this") 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){
return (expectedValue == compareAndSwap(expectedValue, newValue));
}
}

　　当多个线程尝试使用CAS同时更新同一个变量时，只有其中一个线程能更新变量的值，而其他线程都将失败。然而，失败的线程并不会被挂起（这与获取锁的情况不同：当获取锁失败时，线程将被挂起），而是告知子啊这次竞争中失败，并可以再次尝试。由于一个线程在竞争CAS时失败不会阻塞，因此它可以决定是否重新尝试，或者执行一些恢复操作，也可以不执行任何操作。这种灵活性就大大减少了与锁相关的活跃性风险。

　　CAS的典型使用模式是：首先从V中读取值A，并根据A计算新值B，然后再通过CAS以原子方式将V中的值由A变成B（只要在这期间没有任何线程将V的值修改为其他值）。由于CAS能检测到来自其他线程的干扰，因此即使不使用锁也能够实现原子的 读——改——写 操作序列。

非阻塞的计数器

　　下面程序中的CasCounter 使用了CAS实现了一个线程安全的计数器。递增采用了标准形式——读取旧的值，根据它计算出新值（加1），并使用CAS来设置这个新值。如果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;
}
}

CAS的主要缺点是：它将使调度者处理竞争问题（通过重试、回退、放弃），而在使用锁中能自动处理竞争问题（线程在获得锁之前将一直阻塞）。

JVM对CAS的支持

java.util.concurrent.atomic 类的小工具包，支持在单个变量上解除锁的线程安全编程。
AtomicBoolean 可以用原子方式更新的 boolean 值。
AtomicInteger 可以用原子方式更新的 int 值。
AtomicIntegerArray 可以用原子方式更新其元素的 int 数组。
AtomicIntegerFieldUpdater<T> 基于反射的实用工具，可以对指定类的指定 volatile int 字段进行原子更新。
AtomicLong 可以用原子方式更新的 long 值。
AtomicLongArray 可以用原子方式更新其元素的 long 数组。
AtomicLongFieldUpdater<T> 基于反射的实用工具，可以对指定类的指定 volatile long 字段进行原子更新。
AtomicMarkableReference<V> AtomicMarkableReference 维护带有标记位的对象引用，可以原子方式对其进行更新。
AtomicReference<V> 可以用原子方式更新的对象引用。
AtomicReferenceArray<E> 可以用原子方式更新其元素的对象引用数组。
AtomicReferenceFieldUpdater<T,V> 基于反射的实用工具，可以对指定类的指定 volatile 字段进行原子更新。
AtomicStampedReference<V> AtomicStampedReference 维护带有整数“标志”的对象引用，可以用原子方式对其进行更新。

　　在Java 5.0之前，如果不编写明确的代码，那么就无法执行CAS。在Java 5.0引入了底层的支持，在int 、long 和对象的引用等类型上都公开了CAS操作，并且JVM把他们编译为底层硬件提供的最有效方法。在支持CAS的平台上，运行时把它们编译为相应的（多条）机器指令。在最坏的情况下，如果不支持CAS指令，那么JVM将用自旋锁。在原子变量类（例如 java.util.concurrent.atomic中的AtomicXxx）中使用了这些底层的JVM支持为数字类型和引用类型提供一种高效的CAS操作，而在java.util.concurrent 中大多数类在现实时则直接或者间接地使用了这些原子变量类。

第3部分 原子变量类

　　原子变量比锁的粒度更细，量级更轻，并且对于在多处理器系统上实现高性能的并发代码来说是非常关键的。原子变量将发生竞争的范围缩小到单个变量上，这是你获得的粒度最新的情况。

原子变量是一种“更好的volatile”

通过CAS来维持包含多个变量的不变性条件例子：在下面程序中CasNumberRange 中使用了AtomicReference和IntPair 来保存状态，并通过使用Compare-AndSet,使它在更新上界或下界时能避免NumberRange的竞态条件。

import java.util.concurrent.atomic.AtomicReference;

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

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( "Cant't set lower to " + i + " > upper");
}
IntPair newv = new IntPair(i, oldv.upper );
if (values .compareAndSet(oldv, newv)){
return ;
}
}
}
// 对setUpper采用类似的方法
}

性能比较：锁与原子变量

使用ReentrantLock、AtomicInteger、ThreadLocal比较，通常情况下效率排序是ThreadLocal > AtomicInteger > ReentrantLock。

第4部分 非阻塞算法

如果在某种算法中，一个线程的失败或挂起不会导致其他线程也失败或挂起，那么这种算法就被称为非阻塞算法。

如果在算法的每个步骤中都存在某个线程能够执行下去，那么这种算法也被称为无锁（Lock-Free）算法。

到目前为止，已经看到了一个非阻塞算法：CasCounter。在许多常见的数据结构中都可以使用非阻塞算法，包括栈、队列、优先队列以及散列表等。

4.1 非阻塞的栈

　　创建非阻塞算法的关键在于，找出如何将原子修改的范围缩小到单个变量上，同时还要维护数据的一致性。

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

使用 Treiber 算法的非阻塞堆栈

/**
* ConcurrentStack
* <p/>
* Nonblocking stack using Treiber's algorithm
*
* @author Brian Goetz and Tim Peierls
*/
@ThreadSafe
public class ConcurrentStack <E> {
AtomicReference<Node<E>> top = new AtomicReference<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;
}
}
}

在CasCounter 和 ConcurrentStack中说明了非阻塞算法的所有特性：某项工作的完成具有不确定性，必须重新执行。

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

4.2 非阻塞的链表

CAS的基本使用模式：在更新某个值时存在不确定性，以及在更新失败时重新尝试。构建非阻塞算法的技巧在于：将执行原子修改的范围缩小到单个变量上。

　　链接队列比栈更为复杂，因为它必须支持对头节点和尾节点的快速访问。因此，它需要单独维护的头指针和尾指针。有两个指针指向尾部的节点：当前最后一个元素的next指针，以及尾节点。当成功地插入一个新元素时，这两个指针都需要采用原子操作来更新。

　　这里需要一些技巧来完成，第一个技巧是，即使在一个包含多个步骤的更新操作中，也要确保数据结构总是处于一致的状态。这样，当线程B到达时，如果发现线程A正在执行更新，那么线程B就可以知道有一个操作已部分完成，并且不能立即开始执行自己的更新操作。然后，B可以等待（通过反复检查队列的状态）并直到A完成更新，从而使两个线程不会相互干扰。

　　虽然这种方法能够使不同的线程“轮流”访问呢数据结构，并且不会造成破坏，但如果一个线程在更新操作中失败了，那么其他的线程都无法在访问队列。要使得该算法成为一个非阻塞算法，必须确保当一个线程失败时不会妨碍其他线程继续执行下去。因此，第二个技巧是，如果当B到达时发现A正在修改数据结构，那么在数据结构中应该有足够多的信息，使得B能完成A的更新操作。如果B“帮助”A完成了更新操作，那么B可以执行自己的操作，而不用等到A的操作完成。当A恢复后再试图完成其操作时，会发现B已经替它完成了。

　　在下面的程序中，给出了 Michael-Scott 提出的非阻塞连界队列算法中的插入部分，它是由

ConcurrentLinkedQueue

实现的。在许多队列算法中，空队列通常都包含一个“哨兵节点”或者“哑（Dummy）节点”，并且头节点和尾节点在初始化时都指向该哨兵节点。尾节点通常要么指向哨兵节点（如果队列为空），即队列的最后一个元素，要么（当有操作正在执行更新时）指向倒数第二个元素。下图1给出了一个处于正常状态（或者说稳定状态）的包含两个元素的队列。

Michael-Scott 非阻塞队列算法中的插入：

@ThreadSafe
public class LinkedQueue<E> {
private static class Node <E> {
final E item;
final AtomicReference<LinkedQueue.Node<E>> next;

public Node(E item, LinkedQueue.Node<E> next) {
this.item = item;
this.next = new AtomicReference<LinkedQueue.Node<E>>(next);
}
}

private final LinkedQueue.Node<E> dummy = new LinkedQueue.Node<E>(null, null);
private final AtomicReference<LinkedQueue.Node<E>> head
= new AtomicReference<LinkedQueue.Node<E>>(dummy);
private final AtomicReference<LinkedQueue.Node<E>> tail
= new AtomicReference<LinkedQueue.Node<E>>(dummy);

public boolean put(E item) {
LinkedQueue.Node<E> newNode = new LinkedQueue.Node<E>(item, null);
while (true) {
LinkedQueue.Node<E> curTail = tail.get();
LinkedQueue.Node<E> tailNext = curTail.next.get();
if (curTail == tail.get()) {
if (tailNext != null) {  // A
// 队列处于中间状态，推进尾节点
tail.compareAndSet(curTail, tailNext); // B
} else {
// 处于稳定状态，尝试插入新节点
if (curTail.next.compareAndSet(null, newNode)) { // C
// 插入操作成功，尝试推进尾节点
tail.compareAndSet(curTail, newNode); // D
return true;
}
}
}
}
}
}

　　　　图1 处于稳定状态并包含两个元素的对立

　　当插入一个新的元素时，需要更新两个指针。首先更新当前最后一个元素的next 指针，将新节点链接到队列队尾，然后更新尾节点，将其指向这个新元素。在两个操作之间，队列处于一种中间状态，如图2。在等二次更新完成后，队列将再次处于稳定状态，如图3所示。

　　实现这两个技巧的关键在于：当队列处于稳定状态时，尾节点的next域将为空，如果队列处于中间状态，那么tail.next 将为非空。因此，任何线程都能够通过检查tail.next 来获取队列当前的状态。而且，当队列处于中间状态时，可以通过将尾节点移动一个节点，从而结束其他线程正在执行的插入元素操作，并使得队列恢复为稳定状态。



　　　　图2 在插入过程中处于中间状态的对立



　　　　图3 在插入操作完成后，队列再次处于稳定状态

　　LinkedQueue.put 方法在插入新元素之前，将首先检查队列是否处于中间状态（步骤A）。如果是，那么有另一个线程正在插入元素（在步骤C和D之间）。此时当前线程不会等待其他线程执行完成，而是帮助它完成操作，并将尾节点向前推进一个节点（步骤B）。然后，它将重复执行这种检查，以免另一个线程已经开始插入新元素，并继续推进尾节点，直到它发现队列处于稳定状态之后，才会开始执行自己的插入操作。

　　由于步骤C中的CAS将把新节点链接到队列尾部，因此如果两个线程同时插入元素，那么这个CAS将失败。在这样的情况下，并不会造成破坏：不会发生任何变化，并且当前的线程只需要重新读取尾节点并再次重试。如果步骤C成功了，那么插入操作将生效，第二个CAS（步骤D）被认为是一个“清理操作”，因为它既可以由执行插入操作的线程来执行，也可以由其他任何线程来执行。如果步骤D失败，那么执行插入操作的线程将返回，而不是重新执行CAS，因为不再需要重试——另一个线程已经在步骤B中完成了这个工作。

　　这种方式能够工作，因为在任何线程尝试将一个新节点插入到队列之前，都会首先通过检查tail.next是否非空来判断是否需要清理队列。如果是，它首先会推荐尾节点（可能需要执行多次），直到队列处于稳定状态。

4.3 ABA问题

　　ABA问题是一种异常现象：如果在算法中的节点可以被循环使用，那么在使用“比较并交换”指令时就可能出现这种问题（主要在没有垃圾回收机制的环境中）。在CAS操作中将判断“V的值是否仍然是A?”，并且如果是的话就继续执行更新操作。在大多数情况下，这种判断是完全足够的。然而，有时候还需要知道“自从上次看到V的值为A以来，这个值是否发生了变化？” 在某些算法中，如果V的值首先由A变成B，再由B变成A，那么仍然被认为是发生了变化，并需要重新执行算法中的某些步骤。

　　有一个相对简单的解决方案：不是更新某个引用的值，而是更新两个值，包括一个引用和一个版本号。即使这个值由A变成 B，然后又变成A，版本号也将是不同的。

参考

1.《java并发编程实战》 第15章

2.原子变量与非阻塞算法

3.从Java到CPU指令系列之 - 锁和原子操作是怎么实现的 How Lock and atomic works