高并发原子累加器Striped64及其实现类LongAdder&LongAccumulator

Striped64原理

通过前面的几章关于原子类的同步数据结构分析，我们知道Java并发包提供的原子类都是采用volatile+CAS机制实现的，这种轻量级的实现方式比传统的synchronize一般来说更加高效，但是在高并发下依然会导致CAS操作的大量竞争失败自旋重试，这时候对性能的影响说不定还不如使用synchronize，幸运的是，从JDK8开始Java并发包新增了抽象类Striped64以及它的扩展类 LongAdder、LongAccumulator、DoubleAdder、DoubleAccumulator解决了高并发下的累加问题。

       我们知道普通的原子类例如AtomicLong等，它们内部都维护了一个volatile类型的变量用于保存对应的值，由于所有的操作都最终要通过CAS作用到这个变量上，所以在高并发环境下竞争是无可避免的，而Striped64的原理很简单，Striped64不再使用单个变量保存结果，而是包含一个基础值base和一个单元哈希表cells其实就是一个数组。没有竞争的情况下，要累加的数会累加到这个基础值上；如果有竞争的话，会通过内部的分散计算将要累加的数累加到单元哈希表中的某个单元里面。所以整个Striped64的值包括基础值和单元哈希表中所有单元的值的总和。显然Striped64是一种以空间换时间的解决方案。

      前面关于CPU缓存行“伪共享”和ThreadLocalRandom的分析其实都是为本文铺路，因为这里都会用到这些知识。

 

Striped64源码分析

先看看基本结构：

@SuppressWarnings("serial")
abstract class Striped64 extends Number {

    //存放Cell的哈希表，大小为2的幂
    transient volatile Cell[] cells;

    //基础值, 主要时当没有竞争是直接更新这个值, 但也可以作为哈希表初始化竞争失败的回退方案
    //通过CAS的方式更新
    transient volatile long base;

    //自旋锁（通过CAS方式），用于当需要扩展数组的容量或创建一个数组中的元素时加锁.
    transient volatile int cellsBusy;
    //下面是Cell的结构，就是数组中每个元素的结构
    //这其实就是一个AtomicLong的变种，还使用了Contended注解解决伪共享的问题
    @sun.misc.Contended static final class Cell {
        volatile long value;
        Cell(long x) { value = x; }//构造方法
        final boolean cas(long cmp, long val) {//CAS更新方法
            return UNSAFE.compareAndSwapLong(this, valueOffset, cmp, val);
        }

        // Unsafe mechanics
        private static final sun.misc.Unsafe UNSAFE;
        private static final long valueOffset;
        static {
            try {
                UNSAFE = sun.misc.Unsafe.getUnsafe();
                Class<?> ak = Cell.class;
                valueOffset = UNSAFE.objectFieldOffset
                    (ak.getDeclaredField("value"));
            } catch (Exception e) {
                throw new Error(e);
            }
        }
    }
	....
}

从以上的结构看出，Striped64内部维护了一个基础值base，一个存放高竞争时的分散哈希表，即数组cells，数组存放的元素类型Cell是一个类似AtomicLong的变种也是原子类型，另外还有一个自旋锁标记cellsBusy，只用于对数组进行扩容或者创建一个新元素的时候加锁。另外Cell类被Contened注解解决了伪共享的问题，这是因为数组中的元素更倾向于彼此相邻的存放，因此将可能共享缓存行这将会对性能造成巨大的副作用。

 

Striped64主要提供了longAccumulate和doubleAccumulate方法来支持子类，这两个方法也是Striped64最核心的方法，先看下longAccumulate：

final void longAccumulate(long x, LongBinaryOperator fn, boolean wasUncontended) {
	int h;
	if ((h = getProbe()) == 0) { //获取当前线程的probe值作为hash值。
		ThreadLocalRandom.current(); //如果probe值为0，强制初始化当前线程的probe值，这次初始化的probe值不会为0。
		h = getProbe(); //再次获取probe值作为hash值。  
		wasUncontended = true; //重新计算了hash值之后，将未竞争标识为true
	}
	boolean collide = false;                // True if last slot nonempty
	for (;;) {   //CAS的标志性方式
		Cell[] as; Cell a; int n; long v;
		if ((as = cells) != null && (n = as.length) > 0) {//哈希表已经初始化过了
		    //通过(n - 1) & h 来定位当前线程被分散到的Cell数组中的位置
			if ((a = as[(n - 1) & h]) == null) { //如果当前位置是空
				if (cellsBusy == 0) { //并且自旋锁标记为空闲      
					Cell r = new Cell(x); 
					if (cellsBusy == 0 && casCellsBusy()) {
					    //成功获取自旋锁标记之后，
						boolean created = false;
						try {               
							Cell[] rs; int m, j;
							//再次检查该位置是否为空
							if ((rs = cells) != null &&
								(m = rs.length) > 0 &&
								rs[j = (m - 1) & h] == null) {
								rs[j] = r; //将新建的代表x的cell放到指定位置。 
								created = true;
							}
						} finally {
							cellsBusy = 0;//释放cellsBusy锁。 
						}
						if (created)
							break;   //如果创建成功，直接跳出循环，退出方法。  
						continue;   //说明上面指定的cell的位置上有cell了，继续尝试。 
					}
				}
				collide = false; //走到这里说明获取cellsBusy锁失败
			}
			//到这里说明上面通过h选定的cell表的位置上已经有Cell了，
			else if (!wasUncontended)       // CAS already known to fail
			    //如果之前的CAS失败，说明已经发生竞争，
				//这里会设置未竞争标志位true，然后进入advanceProbe产生新的probe值，然后重试。
				wasUncontended = true;      // Continue after rehash
			else if (a.cas(v = a.value, ((fn == null) ? v + x :
										 fn.applyAsLong(v, x)))) //如果还未发生竞争，则尝试将x累加到该位置（a）上
				//成功加x累加到该位置（a）上，退出方法，结束。						 
				break;
			else if (n >= NCPU || cells != as) //到这里说明该位置不为空，但是尝试累加到该位置上失败，
			    //如果哈希表即数组cells已经到最大或数组发生了变化
				//这里设置冲突collide为false，然后进入advanceProbe产生新的probe值，然后重试。
				collide = false;            // At max size or stale
			else if (!collide)
				collide = true; //设置冲突标志，表示发生了冲突，重试。  
			else 
			//到这里说明该位置不为空，但是尝试累加到该位置上失败，并且数组的容量还未到最大值，数组也没有发生变化，但是发生了冲突
			if (cellsBusy == 0 && casCellsBusy()) { //尝试获取cellsBusy锁。  
			    
				try {
					if (cells == as) {      //再次确认数组无变化
					    //对数组进行扩容
						Cell[] rs = new Cell[n << 1];
						for (int i = 0; i < n; ++i)
							rs[i] = as[i];
						cells = rs;
					}
				} finally {
					cellsBusy = 0;//释放cellsBusy锁。
				}
				collide = false;
				continue; //扩容哈希表后，然后重试。           
			}
			h = advanceProbe(h); //重新计算新的probe值以对应到不同的下标元素，然后重试。
		}else if (cellsBusy == 0 && cells == as && casCellsBusy()) {
		    //哈希表还未创建，尝试获取cellsBusy锁，成功
			boolean init = false;
			try {                           // Initialize table
				if (cells == as) {
				    //初始化哈希表cells，初始容量为2
					Cell[] rs = new Cell[2];
					rs[h & 1] = new Cell(x);//将当前操作数x放进该数组中
					cells = rs;
					init = true;
				}
			} finally {
				cellsBusy = 0;//释放cellsBusy锁
			}
			if (init)
				break; //初始化表成功，退出方法，结束。
		}
		//如果创建哈希表由于竞争导致失败，尝试将x累加到base上。
		else if (casBase(v = base, ((fn == null) ? v + x :
									fn.applyAsLong(v, x)))) // Fall back on using base
			break; //成功累加到base上，退出方法，结束     
	}
}

以上方法的逻辑稍显复杂，我简单概括如下：

    1. 在第一个if语句块中，判断当前线程的哈希探针值是否初始化，只有当前线程第一个进入这个方法的时候才是未初始化的（这种延迟初始化在上一文的ThreadLocalRandom中已经介绍过）， 如果未初始化，执行ThreadLocalRandom.current()，该方法在上一文ThreadLocalRandom的分析中已经分析过了，它只要就是原子的初始化当前线程的哈希探针值和种子，它保证了不同的线程将具有不同的初始值，这也为后面进行取模映射时将不同线程尽量映射到不同的数组下标减少冲突，提高CAS的成功率从而提高并发效率提高了基础。重新获取到新的哈希值h之后，就要进入下面的核心代码块了。这里的wasUncontended表示之前没有发生CAS竞争失败，一般是为了当wasUncontended为false时重新产生哈希值从而重试定位不到不同的数组下标进行累加。这里已经产生了新的哈希值，所以就将wasUncontended设为了true。

    2. 进入 核心代码块 for (;;)，这是CAS机制的标志性使用方式，它的逻辑如下：

1. if 该哈希表即数组已经初始化过了
	1.1 if 映射到的槽位（下标）是空的，即还没有放置过元素
	    1.1.1 if 锁空闲，加锁后再次判断，如果该槽位仍然是空的，初始化cell并放到该槽。成功后退出。
	    1.1.2 锁已经被占用了，设置collide为false，会导致重新产生哈希重试。
	1.2 else if （槽不为空）在槽上之前的CAS已经失败，刷新哈希重试。
	1.3 else if （槽不为空、且之前的CAS没失败，）在此槽的cell上尝试更新，成功退出。
	1.4 else if 表已达到容量上限或被扩容了，刷新哈希重试。
	1.5 else if 如果不存在冲突，则设置为存在冲突，刷新哈希重试。
	1.6 else if 如果成功获取到锁，则扩容。
	1.7 else 刷新哈希值，尝试其他槽。
2. else if （表未初始化）锁空闲，且数组无变化，且成功获取到锁：
	2.1 初始化哈希表的大小为2，根据取模（h & 1）将需累加的参数x放到对应的下标中。释放锁。
3. else if （表未初始化，锁不空）尝试直接在base上更新，成功返回，失败回到步骤1重试。

过程还是比较复杂，但原理其实很简单，Striped64对性能的提升原因重要的点有以下两点：

1. 加锁的时机：只有在初始化表、扩展表空间和在空槽位上放入值的时候才会加锁，其它时候都采用乐观锁CAS直接尝试，失败之后通过advanceProbe方法刷新哈希值之后换到不同的槽位继续尝试，而不是死等。这种方式在高并发下显然是高效的。

2. 数组的容量：初始容量为2，以后每次通过 <<移位运算增加容量，保证容量大小是2的幂，所以可以使用(length - 1) & h这种取模方式来索引，容量大小上限为大于等于CPU核心数，这是因为如果每个线程对应一个CPU核心，将会存在一个完美的哈希函数映射线程到不同的槽位（数组下标），从而可以尽可能的消除多个线程竞争同一个槽位，提高了并发效率。

Striped64中其他方法除了doubleAccumulate()之外都是一些辅助方法，不再描述，而doubleAccumulate方法是针对double值做累加的，逻辑和longAccumulate一致。但由于Cell内部用long保存数据，所以在累加的时候会利用Double的doubleToRawLongBits和longBitsToDouble方法做double和longBits形式的double之间的转换。

 

LongAdder和DoubleAdder

上面对 Striped64的原理进行了分析，下面要理解它的子类就非常简单了，Striped64的子类主要有LongAdder、LongAccumulator、DoubleAdder、DoubleAccumulator，它们常用于状态采集、统计等场景。AtomicLong/AtomicDouble也可以用于这种场景，但在线程竞争激烈的情况下，LongAdder/DoubleAdder要比AtomicLong/AtomicDouble拥有更高的吞吐量，但会耗费更多的内存空间。

 

 

LongAdder很简单，其中它的add方法最重要：

public class LongAdder extends Striped64 implements Serializable {
public LongAdder() { //只有一个无参构造方法
}
public void add(long x) {
	Cell[] as; long b, v; int m; Cell a;
	if ((as = cells) != null || !casBase(b = base, b + x)) { 
		boolean uncontended = true;
		if (as == null || (m = as.length - 1) < 0 ||
			(a = as[getProbe() & m]) == null ||
			!(uncontended = a.cas(v = a.value, v + x)))
			longAccumulate(x, null, uncontended);
	}
}

LongAdder只有一个无参的构造方法，无法像AtomicLong那样设置初始值，所以它的初始值是0，它的add方法会在哈希表未初始化的时候才尝试在base上累加，如果已经初始化了哈希表（但对应的槽位为空或尝试累加到该槽位失败）或者在base上累加失败， 则调用父类Striped64的longAccumulate方法进行分散累加。

• 这里如果是第一次执行getProbe()，返回值肯定为0，因为没有通过ThreadLocalRandom.current（）初始化哈希探针值，所以第一次将会在0槽位进行尝试。
• 如果哈希表初始化之后，以后每一次都不会再在base上尝试累加了，那么能不能将第一个if判断中的条件进行调换呢？比如，先做casBase的判断？结果是不调换可能更好，调换后每次都要CAS一下，在高并发时，失败几率非常高，并且是恶性循环，比起一次判断，后者的开销明显小很多，还没有副作用。因此，不调换可能会更好。

LongAdder的其他方法很简单，列举如下：

public void increment() //累加1
public void decrement() //减1
public long sum() //求和，即base值加上每个cell的值。
public void reset() //重置方法，将base和cells中的元素都置为0。 
public long sumThenReset() //先求和，再重置
下面这几个方法是重写的父类Number的方法。
public long longValue() //求和
public int intValue() //求和之后强转int
public float floatValue() //求和之后强转float
public double doubleValue() //求和之后强转double
还有序列号相关的方法....

DoubleAdder的实现与LongAdder一样， 它是利用的父类Striped64的doubleAccumulate方法，只不过使用了Double.doubleToRawLongBits(double)和Double.longBitsToDouble(long)方法在double和longBits数据之间转换，另外DoubleAdder也没有像increment()/decrement()这种加减1的方法，因为它一般不是操作的整数，也就不必要了。

 

LongAccumulator和DoubleAccumulator以及它们的局限性

LongAccumulator和DoubleAccumulator的构造方法与LongAdder/DoubleAdder不同，LongAdder/DoubleAdder只有一个无参的构造方法，不能指定初始值，而它们的构造方法有两个参数，第一个参数是一个需要被实现累加逻辑的函数接口，第二个参数就是初始值。

    LongAccumulator/DoubleAccumulator的使用有很大的局限性，根据JDK8的doc描述：

JDK8的Doc 写道

The order of accumulation within or across threads is not guaranteed and cannot be depended upon, so this class is only applicable to functions for which the order of accumulation does not matter. The supplied accumulator function should be side-effect-free, since it may be re-applied when attempted updates fail due to contention among threads. The function is applied with the current value as its first argument, and the given update as the second argument.

    也就是说，线程之间的累加顺序无法保证，也不应该被依赖，它们仅仅适用于对累加顺序不敏感的累加操作，构造方法的第一个参数指定的累加函数必须是无副作用的，例如（v*2+x）这样的累加函数就不适用在这里，其实这很好理解，Striped64的思想是分散的将要累加的数据放到一个哈希表里面，当执行（v*2+x）这样的函数时，第一个参数v要么是0（空槽位），要么是base（无竞争），要么是某个槽位中的值（无竞争的CAS某个非空槽位时），v不是当前数据的总和，而是根据线程的不同执行顺序而对应到不同的值，所以它的计算结果也将发生偏差，所以上面的Doc中对于第一个参数的描述也是错误的（它描述的第一个参数是当前值），在分散计算的时候，第一个参数并不是累加器的当前值。

    下面的例子是来至stackoverflow，很好的诠释了这个局限性：

public static void main(String[] args) throws InterruptedException {
	LongBinaryOperator op = (x, y) -> 2 * x + y;
	LongAccumulator accumulator = new LongAccumulator(op, 1L);

	ExecutorService executor = Executors.newFixedThreadPool(2);
        
        //产生【0,9）的10个数字，用两个线程去执行累加这10个数
	IntStream.range(0, 10)
			 .forEach(i -> executor.submit(() -> accumulator.accumulate(i)));

	stop(executor);

	System.out.format("Add: %d\n", accumulator.getThenReset());
}

public static void stop(ExecutorService executor) {
	try {
		executor.shutdown();
		executor.awaitTermination(60, TimeUnit.SECONDS);
	}
	catch (InterruptedException e) {
		System.err.println("termination interrupted");
	}
	finally {
		if (!executor.isTerminated()) {
			System.err.println("killing non-finished tasks");
		}
		executor.shutdownNow();
	}
}

上例中的结果在多次运行情况下，结果是不一致的，一会是2539，一会是2037，这就是因为运算函数 2 * x + y是对顺序敏感的，第一个参数x的值是与线程执行顺序相关的。这个问题被提交为一个JDK的bug，听说在JDK10中被修复了，但经过我查看JDK10的源码，其实JDK10中的LongAccumulator和DoubleAccumulator的逻辑并没有有任何更改，JDK的开发者仅仅是修改了其Java Doc， 更加明确了它的使用局限性，并对第一个参数的不准确描述进行了修正：

JDK10的Doc修正 写道

For predictable results, the accumulator function should be associative and commutative. The function is applied with an existing value (or identity) as one argument, and a given update as the other argument.

对于可预测的结果，累加器函数应该是可交换的，第一个参数是一个存在的中间值或者基础值。由此可见，对 LongAccumulator和DoubleAccumulator的使用有着很大的局限性，这直到JDK10都一样。所以使用的时候一定要注意。

 

LongAdder/DoubleAdder其实是LongAccumulator和DoubleAccumulator的特例，当第一个参数的累加函数式 x+y，第二个参数是0的时候，LongAccumulator和DoubleAccumulator就等价于LongAdder/DoubleAdder。

LongAccumulator和DoubleAccumulator还提供了获取结果（get()，getThenReset）和重置（reset()）等方法，就不再一一介绍了，关键是要对它们的使用局限性要明白，不要乱用。

 

总结

本文首先了分析了Striped64的分散计算方式解决了高竞争的累加问题，然而它是一个抽象类无法直接使用，我们只有使用它的实现类，或自己实现。它的实现类中LongAdder/DoubleAdder是初始值为0，只能进行累加/减1的简单累加器，常用于状态采集、统计等场景。LongAccumulator和DoubleAccumulator虽然比LongAdder/DoubleAdder更加强大，能够指定初始值和计算函数，但是由于其不能依赖执行顺序和必须是无副作用的函数局限性，所以使用起来也必须要非常小心。

 

最后如要问是否可以抛弃AtomicLong、AtomicDouble，直接使用LongAdder/DoubleAdder或LongAccumulator/DoubleAccumulator，我认为不能。首先，其实在非高并发的情况下，它们的执行效率相差不大，但是AtomicLong/AtomicDouble提供的方法更丰富，使用起来更方法，而Striped64的实现类们不但方法少，而且由于解决“伪共享”的问题可能导致空间消耗大。其次，它们的使用场景不一样，AtomicLong/AtomicDouble适用于复杂的细粒度的同步控制，而Striped64的实现类们更多地用于逻辑简单的收集统计数据。