Striped64
这个类，嗯有点难度，我会尽力，在进入真正的内容前有些概念或者知识需要补充，或者说从我的角度认为大家需要知道的，可能很多人都已经知道了。
- Strip(Striping) 中文直译是拆分,条纹化的意思，这个词汇或者这个思想在磁盘IO中应用很多，可以参考着理解这个概念。基本思想是热点数据分离。
源码解析
这个类是包内使用的，不对外，用于为那些支持在64位元素上动态分片的类提供统一的表现和技术实现
这个类继承了Number类，这也就是说具体实现的子类也必须实现相关的内容
这个类持有一个延迟初始化的表格，表格内存储了原子级别的变量更新，外加一个额外的base字段。表的大小是平方的。索引使用每个线程的哈希值掩码后得到。几乎所有这个类里面的定义都是包内可见的，这就意味着，被子类直接访问。
表格内的元素是Cell类，Cell类是一个为了减少缓存争用而填充的AtomicLong的变种（通过 sun.misc.Contended）。此处可能涉及到sun.misc.Contended用于避免伪共享的相关内容。以前都是使用long padding ，jdk 1.8 开始使用sun.misc.Contended来替换。long padding的解决办法不怎么优雅，所以在jdk 1.7中甚至会惨遭“优化”，所以在1.7中程序员们绞尽脑汁避免这个优化。更详细的在关于伪共享的主题中记录。
填充对于大多数原子来说其实是多余的，因为他们通常不规则的分布在内存中，因此彼此之间没有太多的互相干扰。但是驻留在数组中的源自对象通常都彼此相邻，因此，在没有这种预防措施的情况下很有可能出现伪共享，多个元素共享一行缓存，这对性能有着巨大的负面影响。
在某种程度上，因为Cell类比较大，尽可能避免在使用之前创建，也就是说要尽量延后创建。如果没有竞争，那么所有的更新操作将对base字段实现。当发生第一次争用（也就是说如果第一次对base字段的CAS操作失败），这个table就会被初始化为2。当进一步的争用发生的时候,表的大小会增加一倍，直到达到这个数字等于挥着大于cpu的数量，由于是每次2倍，也就是说这个数字是2的n次幂。表应该在需要之前保持为null，也就是未使用的时候应该是是null。
利用一个自旋锁（cellsBusy）来对初始化和table大小修改以及用新Cells填充slots提供锁保证。这个地方没有必要使用阻塞锁，如果锁不可达，线程可以尝试其他的slots，或者 尝试base字段。在这些重试期间，竞争加剧，但是降低了局部性，这仍然比阻塞锁来得好。
线程的标识通过threadLocalRandom维护，实际是每个线程的哈希值。在第一个slot没出现竞争的时候，让它保持为未初始化状态0，出现竞争后初始化为通常不会和其他的的值冲突的值，比如线程的哈希码。在执行更新时发生CAS操作失败意味着出现了争用或者表碰撞，也有可能都有。当发生冲突时，如果表大小小于最大容量，把它的大小增加一倍，除了其他线程持情况。如果哈希后的slot为空，并且锁是可用的，那么创建一些新的Cell。如果存在了那么会进行CAS尝试。通过双重哈希进行重试，利用一个辅助哈希（Marsaglia XorShift随机数算法）来尝试寻找一个空闲的slot。
表的大小是收到限制的，因为当线程数超过了cpu数量，假设每个线程都是绑定到一个cpu的，那么一定会存在一个完美的哈希函数，能够将这些线程映射到slot上，从而消除冲突。当我们达到最大容量时，我们通过随机改变冲突线程的哈希代码来搜索这个映射。搜索是随机的，并且只有通过CAS失败才能知道冲突，所以收敛速度可能很慢，而且由于线程通常不会永远绑定到CPU，所以可能根本不会发生。虽然尽管存在这些局限性，但观察到的争用率在这些情况下仍然较低。
Cell可能会出现不可用的情况，包括进行哈希的线程终止，或者由于table扩容导致线程哈希不正确。不会尝试发现或者移除这些Cell，对于处于长时间运行得实例，争用可能会重复出现，所以这时候可能这些Cell又会需要，没有必要销毁，而对于短时间运行的实例，花费时间去销毁又没有什么必要。
Cell类
Atomiclong的填充变体，仅支持原始访问和CAS
JVM内部注释:这里可能会在提供的情况下使用一个正式的CAS操作。这句话翻译出来好像不太好懂，原始主时中有个release-only form of CAS不太好理解，这句话的注释主要是给JVM实现的，CAS不适唯一的原子操作，这里跟C++有关，如果实在不明白，贴一个StackoverFlow的地址，用于深入研究release-only form of CAS
//这个注解会交个jvm进行合适的填充
//关于这个注解
@sun.misc.Contended static final class Cell {
//关于volatile此处不会展开了
//这个变量是实际承载这个真实值的
volatile long value;
//构造方法
Cell(long x) { value = x; }
//执行CAS操作
final boolean cas(long cmp, long val) {
//具体实现的CAS使用了UNSAFE，具体在UNSAFE中记录
return UNSAFE.compareAndSwapLong(this, valueOffset, cmp, val);
}
// Unsafe mechanics
//就像UnSafe字面意义一样，减少使用这个类，毕竟它等同于使用了c的指针
private static final sun.misc.Unsafe UNSAFE;
//
private static final long valueOffset;
//静态初始化块，用于初始化UNSAFE和valueOffset
static {
try {
//Unsafe使用了单例模式
//所以用getUnsafe获取实例。
//Unsafe做了限制，
//如果是一个普通的调用getUnsafe会报SecurityException异常
//只有主类加载器才能调用这个方法。
UNSAFE = sun.misc.Unsafe.getUnsafe();
Class<?> ak = Cell.class;
//计算value元素的偏移量是多少
valueOffset = UNSAFE.objectFieldOffset
(ak.getDeclaredField("value"));
} catch (Exception e) {
throw new Error(e);
}
}
}
接着是一些变量
/** CPU的核心数，和table的大小有关*/
static final int NCPU = Runtime.getRuntime().availableProcessors();
/**
* 用来存储Cell的table，在非空情况下一定是2的幂长度
*/
transient volatile Cell[] cells;
/**
* base字段, 主要在非竞争情况下使用, 也同时充当table在初始化期间的竞争后备
*通过CAS操作进行更新.
*/
transient volatile long base;
/**
* 自旋锁 (通过CAS操作加锁)用于在对Cells做创建或者调整大小的时候进行加锁。
*/
transient volatile int cellsBusy;
/**
* 默认函数使用default修饰，也就是包内有效。
*/
Striped64() {
}
一些方法
/**
* 通过CAS操作，修改Base字段
*/
final boolean casBase(long cmp, long val) {
return UNSAFE.compareAndSwapLong(this, BASE, cmp, val);
}
/**
* 通过CAS操作修改cellsBusy字段，从0修改为1，标示获得锁。
*/
final boolean casCellsBusy() {
return UNSAFE.compareAndSwapInt(this, CELLSBUSY, 0, 1);
}
/**
* 返回当前线程的标示.
* 由于包限制，这段代码是从ThreadLocalRandom拷贝过来的
* Duplicated from ThreadLocalRandom because of packaging restrictions.
*/
static final int getProbe() {
return UNSAFE.getInt(Thread.currentThread(), PROBE);
}
/**
* 利用伪随机算法加强标识后，将为当前线程记录这个标识。
* 由于包限制，这段代码是从ThreadLocalRandom拷贝过来的
* Duplicated from ThreadLocalRandom because of packaging restrictions.
*/
static final int advanceProbe(int probe) {
probe ^= probe << 13; // xorshift
probe ^= probe >>> 17;
probe ^= probe << 5;
UNSAFE.putInt(Thread.currentThread(), PROBE, probe);
return probe;
}
接下来是最大的重头戏，很复杂的两个方法，尽力讲解
这个方法处理初始化、调整大小、创建新Cells和争用等情况。
这个方法由于有比较乐观的重试机制，所以存在常见的非模块化问题，依赖于重试
//x 元素
//fn 更新函数，如果是add可以为null，
//这个约定避免了longadder中定义额外的变量或者函数
//wasUncontended 如果CAS在调用之前失败了，这个值为false
final void longAccumulate(long x, LongBinaryOperator fn,
boolean wasUncontended) {
int h;
//获取当前线程的probe值，如果为0，则需要初始化该线程的probe值
if ((h = getProbe()) == 0) {
// 强制初始化
ThreadLocalRandom.current();
h = getProbe();
wasUncontended = true;
}
//如果上一个slot不为空置为true
//碰撞标记
boolean collide = false;
for (;;) {
Cell[] as; Cell a; int n; long v;
//Cells不为空，进行操作
if ((as = cells) != null && (n = as.length) > 0) {
//通过（hashCode & (length - 1)）这种算法来实现取模
//如果当前位置为null说明需要初始化
if ((a = as[(n - 1) & h]) == null) {
//判断锁状态
//尝试添加新的cell
if (cellsBusy == 0) { // Try to attach new Cell
Cell r = new Cell(x); // Optimistically create
//再次判断锁状态，同时获取锁
if (cellsBusy == 0 && casCellsBusy()) {
boolean created = false;
//利用try确保处于获取锁的状态下
try { // Recheck under lock
Cell[] rs; int m, j;
if ((rs = cells) != null &&
(m = rs.length) > 0 &&
rs[j = (m - 1) & h] == null) {
//Cell添加
rs[j] = r;
//标示创建
created = true;
}
} finally {
//释放锁
cellsBusy = 0;
}
//创建成功跳出，否则重试
if (created)
break;
continue; // Slot is now non-empty
}
}
collide = false;
}
//运行到此说明cell的对应位置上已经有相应的Cell了，
//不需要初始化了
//CAS操作已经失败了，出现了竞争
else if (!wasUncontended) // CAS already known to fail
wasUncontended = true; // Continue after rehash
//尝试去修改a上的计数，
//a为Cell数组中index位置上的cell
else if (a.cas(v = a.value, ((fn == null) ? v + x :
fn.applyAsLong(v, x))))
break;
//cell数组最大为cpu的数量，
//cells != as表明cells数组已经被更新了
//标记为最大状态或者说是过期状态
else if (n >= NCPU || cells != as)
collide = false; // At max size or stale
else if (!collide)
collide = true;
//尝试获取锁之后扩大Cells
else if (cellsBusy == 0 && casCellsBusy()) {
try {
if (cells == as) { // Expand table unless stale
//Cell数组扩容，每次扩容为原来的两倍
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
}
h = advanceProbe(h);
}
//此分支表明Cells是空的，所以要获取锁，然后初始化Cells
else if (cellsBusy == 0 && cells == as && casCellsBusy()) {
boolean init = false;
try { // Initialize table
if (cells == as) {
Cell[] rs = new Cell[2];
rs[h & 1] = new Cell(x);
cells = rs;
init = true;
}
} finally {
cellsBusy = 0;
}
if (init)
break;
}
//此处表明Cells为空，并且初始化的时候获取锁失败，直接在base上进行CAS
else if (casBase(v = base, ((fn == null) ? v + x :
fn.applyAsLong(v, x))))
break; // Fall back on using base
}
}
这一段的核心是这样的：
- longAccumulate会根据当前线程来计算一个哈希值，然后根据算法(hashCode & (length - 1))来达到取模的效果以定位到该线程被分散到的Cell数组中的位置
- 如果Cell数组还没有被创建，那么就去获取cellBusy这个锁（相当于锁，但是更为轻量级），如果获取成功，则初始化Cell数组，初始容量为2，初始化完成之后将x包装成一个Cell，哈希计算之后分散到相应的index上。如果获取cellBusy失败，那么会试图将x累计到base上，更新失败会重新尝试直到成功。
- 如果Cell数组已经被初始化过了，那么就根据线程的哈希值分散到一个Cell数组元素上，获取这个位置上的Cell并且赋值给变量a，如果a为null，说明该位置还没有被初始化，那么就初始化，当然在初始化之前需要竞争cellBusy变量。
- 如果Cell数组的大小已经最大了（大于等于CPU的数量），那么就需要重新计算哈希，来重新分散当前线程到另外一个Cell位置上再走一遍该方法的逻辑，否则就需要对Cell数组进行扩容，然后将原来的计数内容迁移过去。由于Cell里面保存的是计数值，所以扩容后没有必要做其他处理，直接根据index将旧的Cell数组内容复制到新的Cell数组中。
下面的方法和上面的一样，只不过加入了long和double的转换，如果想要和Long的合并到一起需要在太多的地方加入这种转换，可能会带来额外的开销，为了减少开销，只好通过复制粘贴过来进行调整和维护 。
final void doubleAccumulate(double x, DoubleBinaryOperator fn,
boolean wasUncontended) {
int h;
if ((h = getProbe()) == 0) {
ThreadLocalRandom.current(); // force initialization
h = getProbe();
wasUncontended = true;
}
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) { // Try to attach new Cell
Cell r = new Cell(Double.doubleToRawLongBits(x));
if (cellsBusy == 0 && casCellsBusy()) {
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 is now non-empty
}
}
collide = false;
}
else if (!wasUncontended) // CAS already known to fail
wasUncontended = true; // Continue after rehash
else if (a.cas(v = a.value,
((fn == null) ?
Double.doubleToRawLongBits
(Double.longBitsToDouble(v) + x) :
Double.doubleToRawLongBits
(fn.applyAsDouble
(Double.longBitsToDouble(v), x)))))
break;
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) { // Expand table unless stale
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
}
h = advanceProbe(h);
}
else if (cellsBusy == 0 && cells == as && casCellsBusy()) {
boolean init = false;
try { // Initialize table
if (cells == as) {
Cell[] rs = new Cell[2];
rs[h & 1] = new Cell(Double.doubleToRawLongBits(x));
cells = rs;
init = true;
}
} finally {
cellsBusy = 0;
}
if (init)
break;
}
else if (casBase(v = base,
((fn == null) ?
Double.doubleToRawLongBits
(Double.longBitsToDouble(v) + x) :
Double.doubleToRawLongBits
(fn.applyAsDouble
(Double.longBitsToDouble(v), x)))))
break; // Fall back on using base
}
}
下面是Unsafe的相关结构
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;
PROBE = UNSAFE.objectFieldOffset
(tk.getDeclaredField("threadLocalRandomProbe"));
} catch (Exception e) {
throw new Error(e);
}
}
网友评论