striped64

简介

striped64是java8用来并发计数新加的组件，在并发环境下有更好的性能，64指的是计数64bit的数，即Long类型的数和Double类型的数。striped64是个抽象类，jdk提供了四个实现类，LongAdder LongAccumulator DoubleAdder DoubleAccumulator 我们拿数据说话，分别在并发情况下用synchronized,atomic,striped64进行基准测试。

代码

public class Striped64BenchMark {private static final Object lock = new Object();private static long add;private static AtomicLong add2 = new AtomicLong();private static LongAdder add3 = new LongAdder();public static void main(String[] args) throws InterruptedException {CyclicBarrier begin = new CyclicBarrier(10);CountDownLatch end = new CountDownLatch(10);long maxSum = 100000000;ExecutorService executorService = Executors.newFixedThreadPool(10);/*synchronized*/AtomicLong beginTime = new AtomicLong();for (int j = 0; j < 10; j++) {executorService.execute(() -> {try {synchronized (begin) {if (begin.getNumberWaiting() == 0) {beginTime.set(System.currentTimeMillis());}}begin.await();long i = 0;do {synchronized (lock) {add++;}}while (++i < maxSum);end.countDown();} catch (InterruptedException e) {e.printStackTrace();} catch (BrokenBarrierException e) {e.printStackTrace();}});}end.await();System.out.println("add1 use time is:" + (System.currentTimeMillis() - beginTime.get()));/*atomic*/begin.reset();AtomicLong beginTime2 = new AtomicLong();CountDownLatch end2 = new CountDownLatch(10);for (int j = 0; j < 10; j++) {executorService.execute(() -> {try {synchronized (begin) {if (begin.getNumberWaiting() == 0) {beginTime2.set(System.currentTimeMillis());}}begin.await();long i = 0;do {add2.getAndIncrement();}while (++i < maxSum);end2.countDown();} catch (InterruptedException e) {e.printStackTrace();} catch (BrokenBarrierException e) {e.printStackTrace();}});}end2.await();System.out.println("add2 use time is:" + (System.currentTimeMillis() - beginTime2.get()));/*striped64*/begin.reset();AtomicLong beginTime3 = new AtomicLong();CountDownLatch end3 = new CountDownLatch(10);for (int j = 0; j < 10; j++) {executorService.execute(() -> {try {synchronized (begin) {if (begin.getNumberWaiting() == 0) {beginTime3.set(System.currentTimeMillis());}}begin.await();long i = 0;do {add3.increment();}while (++i < maxSum);end3.countDown();} catch (InterruptedException e) {e.printStackTrace();} catch (BrokenBarrierException e) {e.printStackTrace();}});}end3.await();System.out.println("add3 use time is:" + (System.currentTimeMillis() - beginTime3.get()));}
}

运行结果

add1 use time is:37309
add2 use time is:17996
add3 use time is:4525

设计思想

性能差距还是很明显的，为甚striped64这么快？要从striped64设计思想说起。
striped64采用了类似分段锁的机制，内部数据结构一个long类型的base变量，和持有long类型变量的Cell[]数组。单线程情况下，只需要对base做加减计数，当遇到多线程的时候，计算hash值，找到对应的Cell,用cas来改变Cell中的值，最后的所求的结果就是base+所有Cell中的值，原理相对简单，但代码还是有很多需要关注的点。

接下来撸一撸代码，先来看一看Cell的数据结构。

@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 mechanicsprivate 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);}}}

结构比较简单，只含有了一个volatile的关键字修饰的long型的值，其他都是cas相关的东西，可以不用太关注。需要注意的是@sun.misc.Contended这个注解，代表会对这个类进行缓存行填充，关于缓存行简单介绍一下。

如图:

在核心1上运行的线程想更新变量X，同时核心2上的线程想要更新变量Y。不幸的是，这两个变量在同一个缓存行中。每个线程都要去竞争缓存行的所有权来更新变量。如果核心1获得了所有权，缓存子系统将会使核心2中对应的缓存行失效。当核心2获得了所有权然后执行更新操作，核心1就要使自己对应的缓存行失效。这会来来回回的经过L3缓存，大大影响了性能。如果互相竞争的核心位于不同的插槽，就要额外横跨插槽连接，问题可能更加严重。

另外除了base,cells两个变量外，还有一个变量用来存储扩容或者初始化的时候是否有别的线程竞争cellsBusy

核心方法

final void longAccumulate(long x, LongBinaryOperator fn,boolean wasUncontended) {int h;// 初始化if ((h = getProbe()) == 0) {ThreadLocalRandom.current(); h = getProbe();wasUncontended = true;}// 最后的桶不为空则 true，也用于控制扩容，false重试。boolean collide = false;for (;;) {Cell[] as; Cell a; int n; long v;//已经初始化if ((as = cells) != null && (n = as.length) > 0) {// 对应桶是空的if ((a = as[(n - 1) & h]) == null) {if (cellsBusy == 0) {       // 尝试关联新的Cell// 创建cellCell r = new Cell(x);//double checkif (cellsBusy == 0 && casCellsBusy()) {//创建cellboolean 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;         created = true;}} finally {cellsBusy = 0;     // 释放锁}if (created)break;               // 成功退出continue;           // 已经有别的线程进入了}}// 锁被占用了，扩容或者重hash进入下轮循环collide = false;}// 如何传入的wasUncontended，重hash，进入下轮循环else if (!wasUncontended)       wasUncontended = true;      // 尝试更新当前桶，成功直接退出else if (a.cas(v = a.value, ((fn == null) ? v + x :fn.applyAsLong(v, x))))break;// 表大小达到上限或扩容了；// 最后重hash,进入下轮循环else if (n >= NCPU || cells != 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;                   // 在扩容后的表上重试}// 没法获取锁，重散列，尝试其他桶h = advanceProbe(h);}//初始celllelse if (cellsBusy == 0 && cells == as && casCellsBusy()) {// 加锁的情况下初始化表boolean init = false;try {                           // Initialize tableif (cells == as) {Cell[] rs = new Cell[2];rs[h & 1] = new Cell(x);cells = rs;init = true;}} finally {cellsBusy = 0;     // 释放锁}if (init)break;     // 成功初始化，已更新，跳出循环}//有其他线程在初始化cell,尝试更新base的值else if (casBase(v = base, ((fn == null) ? v + x :fn.applyAsLong(v, x))));break;                          }
}

实现类

以LongAdder为例

public void add(long x) {Cell[] as; long b, v; int m; Cell a;//先判断是cells为空，不为空的概率可能大点，而且耗时操作较小，所以先判断，如果为空，表示单线程计数，尝试进行cas操作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);}
}

获取值方法

//此方法或者的值可能会有误差，加入统计的时候依然进行计数，最后结果就会不精确。
public long 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;
}

JUC之striped64相关推荐

JUC Striped64
java.util.concurrent 该包中提供了很多并发环境中表现优秀的类,今天先来介绍下java.util.concurrent.atomic.Striped64这个类 Striped64的基 ...
【JUC源码专题】Striped64 核心源码分析（JDK8）
文章目录核心变量缓存行填充 longAccumulate 方法方法概览 cells 数组已初始化重新计算随机数扩容前置条件 cells 数组未初始化 cas 更新 Base Striped6 ...
Jdk1.8 JUC源码增量解析(1)-atomic-Striped64
转载自 Jdk1.8 JUC源码增量解析(1)-atomic-Striped64 功能简介: Striped64是jdk1.8提供的用于支持如Long累加器,Double累加器这样机制的基础类. S ...
Jdk1.8 JUC源码增量解析(2)-atomic-LongAdder和LongAccumulator
转载自 Jdk1.8 JUC源码增量解析(2)-atomic-LongAdder和LongAccumulator 功能简介: LongAdder是jdk1.8提供的累加器,基于Striped64实现. ...
Striped64 深入源码解析
JUC在JDK1.8版本中引入了Striped64类及其四个实现类LongAdder.LongAccumulator.DoubleAdder.DoubleAccumulator,在原有的Atomi ...
高并发累加器 Striped64
简介: Striped64 是一个高并发累加器,是一个abstract class (包局部抽象类,同一个包才能访问),只能通过它的子类使用(LongAdder,LongAccumulator,Dob ...
JAVA社交平台项目第七天 JUC多线程
JUC多线程 (一) 学习目标: 掌握多线程的创建掌握线程安全的处理了解线程状态掌握线程停止的两种方法了解线程的原子性,可见性和有序性理解内存可见性的原理掌握synchronized解决内 ...
并发编程JUC深度学习(七）无锁（乐观锁）
前言乐观锁总是假设最好的情况,每次去拿数据的时候都认为数据不会被修改,所以不会上锁,只是在更新的时候会判断一下在这期间别人有没有去更新这个数据,主要是由版本号机制和CAS算法实现,这里我们主要介绍C ...
17万字 JUC 看这一篇就够了(一) (精华)
JUC 今天我们来进入到 Java并发编程 JUC 框架的学习 ,内容比较多,但希望我们都能静下心来,耐心的看完这篇文章文章目录 JUC 进程概述对比线程创建线程 Thread Runnab ...

JUC之striped64

striped64

简介

代码

设计思想

核心方法

实现类

JUC之striped64相关推荐

最新文章

热门文章