内容简介：Java 并发工具类之 LongAdder 原理总结
java.util.concurrency.atomic.LongAdder 是 Java 8新增的一个类，提供了原子累计值的方法。根据文档的描述其性能要优于 AtomicLong ，下图是一个简单的测试对比(平台:MBP):
这里测试时基于JDK1.8进行的，AtomicLong 从Java8开始针对x86平台进行了优化，使用XADD替换了CAS操作，我们知道JUC下面提供的原子类都是基于Unsafe类实现的，并由Unsafe来提供CAS的能力。CAS (compare-and-swap)本质上是由现代CPU在硬件级实现的原子指令，允许进行无阻塞，多线程的数据操作同时兼顾了安全性以及效率。大部分情况下,CAS都能够提供不错的性能，但是在高竞争的情况下开销可能会成倍增长，具体的研究可以参考这篇 文章 , 我们直接看下代码:
public class AtomicLong {
public final long incrementAndGet() {
return unsafe.getAndAddLong(this, valueOffset, 1L) + 1L;
}
}
public final class Unsafe {
public final long getAndAddLong(Object var1, long var2, long var4) {
long var6;
do {
var6 = this.getLongVolatile(var1, var2);
} while(!this.compareAndSwapLong(var1, var2, var6, var6 + var4));
return var6;
}
}
getAndAddLong 方法会以volatile的语义去读需要自增的域的最新值，然后通过CAS去尝试更新，正常情况下会直接成功后返回，但是在高并发下可能会同时有很多线程同时尝试这个过程，也就是说线程A读到的最新值可能实际已经过期了，因此需要在while循环中不断的重试，造成很多不必要的开销，而xadd的相对来说会更高效一点，伪码如下,最重要的是下面这段代码是原子的,也就是说其他线程不能打断它的执行或者看到中间值，这条指令是在硬件级直接支持的:
function FetchAndAdd(address location, int inc) {
int value := *location
*location := value + inc
return value
}
而LongAdder的性能比上面那种还要好很多，于是就研究了一下。首先它有一个基础的值base，在发生竞争的情况下，会有一个Cell数组用于将不同线程的操作离散到不同的节点上去(会根据需要扩容，最大为CPU核数)， sum() 会将所有Cell数组中的value和base累加作为返回值。核心的思想就是将AtomicLong一个value的更新压力分散到多个value中去，从而降级更新热点。
public class LongAdder extends Striped64 implements Serializable {
//...
}
LongAdder继承自 Striped64 ， Striped64 内部维护了一个懒加载的数组以及一个额外的 base 实力域，数组的大小是2的N次方，使用每个线程 Thread 内部的哈希值访问。
abstract class Striped64 extends Number {
/** Number of CPUS, to place bound on table size */
static final int NCPU = Runtime.getRuntime().availableProcessors();
/**
* Table of cells. When non-null, size is a power of 2.
*/
transient volatile Cell[] cells;
@sun.misc.Contended static final class Cell {
volatile long value;
Cell(long x) { value = x; }
final boolean cas(long cmp, long val) {
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);
}
}
}
}
数组的元素是 Cell 类，可以看到Cell类用Contended注解修饰，这里主要是解决false sharing(伪共享的问题)，不过个人认为伪共享翻译的不是很好
static final class Cell {
volatile long p0, p1, p2, p3, p4, p5, p6;
volatile long value;
volatile long q0, q1, q2, q3, q4, q5, q6;
Cell(long x) { value = x; }
final boolean cas(long cmp, long val) {
return UNSAFE.compareAndSwapLong(this, valueOffset, cmp, val);
}
// Unsafe mechanics
private static final sun.misc.Unsafe UNSAFE;
private static final long valueOffset;
static {
try {
UNSAFE = getUnsafe();
Class<?> ak = Cell.class;
valueOffset = UNSAFE.objectFieldOffset
(ak.getDeclaredField("value"));
} catch (Exception e) {
throw new Error(e);
}
}
}
但是这种方式毕竟不通用，例如32、64位操作系统的缓存行大小不一样，因此JAVA8中就增加了一个注 @sun.misc.Contended 解用于解决这个问题,由JVM去插入这些变量，具体可以参考openjdk.java.net/jeps/142 ，但是通常来说对象通常是不规则的分配到内存中的，但是数组由于是连续的内存，因此可能会共享缓存行，因此这里加一个Contended注解以防cells数组发生伪共享的情况。
/**
* 底竞争下直接更新base，类似AtomicLong
* 高并发下，会将每个线程的操作hash到不同的
* cells数组中，从而将AtomicLong中更新
* 一个value的行为优化之后，分散到多个value中
* 从而降低更新热点，而需要得到当前值的时候，直接
* 将所有cell中的value与base相加即可，但是跟
* AtomicLong(compare and change -> xadd)的CAS不同，
* incrementAndGet操作及其变种
* 可以返回更新后的值，而LongAdder返回的是void
*/
public class LongAdder {
public void add(long x) {
Cell[] as; long b, v; int m; Cell a;
/**
* 如果是第一次执行，则直接case操作base
*/
if ((as = cells) != null || !casBase(b = base, b + x)) {
boolean uncontended = true;
/**
* as数组为空(null或者size为0)
* 或者当前线程取模as数组大小为空
* 或者cas更新Cell失败
*/
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);
}
}
public long sum() {
//通过累加base与cells数组中的value从而获得sum
Cell[] as = cells; Cell a;
long sum = base;
if (as != null) {
for (int i = 0; i < as.length; ++i) {
if ((a = as[i]) != null)
sum += a.value;
}
}
return sum;
}
}
/**
* openjdk.java.net/jeps/142
*/
@sun.misc.Contended static final class Cell {
volatile long value;
Cell(long x) { value = x; }
final boolean cas(long cmp, long val) {
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);
}
}
}
abstract class Striped64 extends Number {
final void longAccumulate(long x, LongBinaryOperator fn,
boolean wasUncontended) {
int h;
if ((h = getProbe()) == 0) {
/**
* 若getProbe为0，说明需要初始化
*/
ThreadLocalRandom.current(); // force initialization
h = getProbe();
wasUncontended = true;
}
boolean collide = false; // True if last slot nonempty
/**
* 失败重试
*/
for (;;) {
Cell[] as; Cell a; int n; long v;
if ((as = cells) != null && (n = as.length) > 0) {
/**
* 若as数组已经初始化,(n-1) & h 即为取模操作，相对 % 效率要更高
*/
if ((a = as[(n - 1) & h]) == null) {
if (cellsBusy == 0) { // Try to attach new Cell
Cell r = new Cell(x); // Optimistically create
if (cellsBusy == 0 && casCellsBusy()) {//这里casCellsBusy的作用其实就是一个spin lock
//可能会有多个线程执行了`Cell r = new Cell(x);`,
//因此这里进行cas操作，避免线程安全的问题，同时前面在判断一次
//避免正在初始化的时其他线程再进行额外的cas操作
boolean created = false;
try { // Recheck under lock
Cell[] rs; int m, j;
//重新检查一下是否已经创建成功了
if ((rs = cells) != null &&
(m = rs.length) > 0 &&
rs[j = (m - 1) & h] == null) {
rs[j] = r;
created = true;
}
} finally {
cellsBusy = 0;
}
if (created)
break;
continue; // Slot 现在是非空了，continue到下次循环重试
}
}
collide = false;
}
else if (!wasUncontended) // CAS already known to fail
wasUncontended = true; // Continue after rehash
else if (a.cas(v = a.value, ((fn == null) ? v + x :
fn.applyAsLong(v, x))))
break;//若cas更新成功则跳出循环，否则继续重试
else if (n >= NCPU || cells != as) // 最大只能扩容到CPU数目， 或者是已经扩容成功，这里只有的本地引用as已经过期了
collide = false; // At max size or stale
else if (!collide)
collide = true;
else if (cellsBusy == 0 && casCellsBusy()) {
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;
}
collide = false;
continue; // Retry with expanded table
}
//重新计算hash(异或)从而尝试找到下一个空的slot
h = advanceProbe(h);
}
else if (cellsBusy == 0 && cells == as && casCellsBusy()) {
boolean init = false;
try { // Initialize table
if (cells == as) {
/**
* 默认size为2
*/
Cell[] rs = new Cell[2];
rs[h & 1] = new Cell(x);
cells = rs;
init = true;
}
} finally {
cellsBusy = 0;
}
if (init)
break;
}
else if (casBase(v = base, ((fn == null) ? v + x : // 若已经有另一个线程在初始化，那么尝试直接更新base
fn.applyAsLong(v, x))))
break; // Fall back on using base
}
}
final boolean casCellsBusy() {
return UNSAFE.compareAndSwapInt(this, CELLSBUSY, 0, 1);
}
static final int getProbe() {
/**
* 通过Unsafe获取Thread中threadLocalRandomProbe的值
*/
return UNSAFE.getInt(Thread.currentThread(), PROBE);
}
// Unsafe mechanics
private static final sun.misc.Unsafe UNSAFE;
private static final long BASE;
private static final long CELLSBUSY;
private static final long PROBE;
static {
try {
UNSAFE = sun.misc.Unsafe.getUnsafe();
Class<?> sk = Striped64.class;
BASE = UNSAFE.objectFieldOffset
(sk.getDeclaredField("base"));
CELLSBUSY = UNSAFE.objectFieldOffset
(sk.getDeclaredField("cellsBusy"));
Class<?> tk = Thread.class;
//返回Field在内存中相对于对象内存地址的偏移量
PROBE = UNSAFE.objectFieldOffset
(tk.getDeclaredField("threadLocalRandomProbe"));
} catch (Exception e) {
throw new Error(e);
}
}
}
由于Cell相对来说比较占内存，因此这里采用懒加载的方式，在无竞争的情况下直接更新base域，在第一次发生竞争的时候(CAS失败)就会创建一个大小为2的cells数组，每次扩容都是加倍，只到达到CPU核数。同时我们知道扩容数组等行为需要只能有一个线程同时执行，因此需要一个锁，这里通过CAS更新cellsBusy来实现一个简单的spin lock。
数组访问索引是通过Thread里的threadLocalRandomProbe域取模实现的，这个域是ThreadLocalRandom更新的，cells的数组大小被限制为CPU的核数，因为即使有超过核数个线程去更新，但是每个线程也只会和一个CPU绑定，更新的时候顶多会有cpu核数个线程，因此我们只需要通过hash将不同线程的更新行为离散到不同的slot即可。
我们知道线程、线程池会被关闭或销毁，这个时候可能这个线程之前占用的slot就会变成没人用的，但我们也不能清除掉，因为一般web应用都是长时间运行的，线程通常也会动态创建、销毁，很可能一段时间后又会被其他线程占用，而对于短时间运行的，例如单元测试，清除掉有啥意义呢？
总结
总的来说，LongAdder从性能上来说要远远好于AtomicLong，一般情况下是可以直接替代AtomicLong使用的，Netty也通过一个接口封装了这两个类，在Java8下直接采用LongAdder，但是AtomicLong的一系列方法不仅仅可以自增，还可以获取更新后的值，如果是例如获取一个全局唯一的ID还是采用AtomicLong会方便一点。
参考链接
- https://blogs.oracle.com/dave/atomic-fetch-and-add-vs-compare-and-swap
- https://en.wikipedia.org/wiki/Compare-and-swap
- http://ashkrit.blogspot.com/2014/02/atomicinteger-java-7-vs-java-8.html
- https://dzone.com/articles/adventures-atomiclong
以上就是本文的全部内容，希望本文的内容对大家的学习或者工作能带来一定的帮助，也希望大家多多支持 码农网
猜你喜欢:
- 高并发原理
- Mysql的并发控制原理
- java 并发之CyclicBarrier 原理
- 并发原理抽丝剥茧，线程本地变量 ThreadLocal 的实现原理
- JAVA并发编程--2.synchronied实现原理
- 深入理解Java并发之synchronized实现原理
本站部分资源来源于网络，本站转载出于传递更多信息之目的，版权归原作者或者来源机构所有，如转载稿涉及版权问题，请联系我们。
HTTP Essentials
Stephen A. Thomas、Stephen Thomas / Wiley / 2001-03-08 / USD 34.99
The first complete reference guide to the essential Web protocol As applications and services converge and Web technologies not only assume HTTP but require developers to manipulate it, it is be......一起来看看 《HTTP Essentials》 这本书的介绍吧!
