CAS存在的题目及在Java中的解决方式 - ToB企服应用市场:ToB评测及商务社交产业平台

public class Demo01 {
// 定义一个共享变量 num
private static int num = 0;
public static void main(String[] args) throws InterruptedException {
// 任务：对 num 进行10000次加操作
Runnable mr = () -> {
for (int i = 0; i < 10000; i++) {
num++; // num++并不是原子操作，就会导致原子性问题的产生
}
};
ArrayList<Thread> ts = new ArrayList<>();
// 同时开辟5个线程执行任务
for (int i = 0; i < 5; i++) {
Thread t = new Thread(mr);
t.start();
ts.add(t);
}
for (Thread t : ts) {
t.join();
}
//因此最终会输出的num < 50000
System.out.println("num = " + num);
}
}

复制代码

public class Demo01 {
public static void main(String[] args) throws InterruptedException {
//
AtomicInteger atomicInteger = new AtomicInteger();
// 任务：自增 10000 次
Runnable mr = () -> {
for (int i = 0; i < 10000; i++) {
atomicInteger.incrementAndGet(); //该自增操作是一个原子性的操作
}
};
ArrayList<Thread> ts = new ArrayList<>();
for (int i = 0; i < 5; i++) {
Thread t = new Thread(mr);
t.start();
ts.add(t);
}
for (Thread t : ts) {
t.join();
}
//由于是原子操作，值将一直会是50000
System.out.println("number = " + atomicInteger.get());
}
}

复制代码

/* AtomicInteger.java */
private volatile int value; // value初始取值为0
public final int incrementAndGet() {
// this：自己 new 好的 atomicInteger对象
// valueOffset：内存偏移量
// 1 ：这个方法时自增操作，因此是1
return unsafe.getAndAddInt(this, valueOffset, 1) + 1;
}

复制代码

/* Unsafe.class */
// var1：上面的this，即atomicInteger对象； var2：valueOffset ； var4：要添加的值
public final int getAndAddInt(Object var1, long var2, int var4) {
// var5 旧的预估值
int var5;
do {
// this 和内存 valueOffset，目的是找出这个 value的当前最新值（旧的预估值）
var5 = this.getIntVolatile(var1, var2);
} while (!this.compareAndSwapInt(var1, var2, var5, var5 + var4));
return var5;
}

复制代码

public void add(long x) {
//as 表示cells引用
//b 表示获取的base值
// v 表示期望值
// m表示cells的数组长度 - 1 （cells长度一定是2的幂）
// a表示当前现成命中的cell单元格
Cell[] as; long b, v; int m; Cell a;
//条件一： true->表示cells已经初始化过了，当前线程应该将数据写入到对应的cell中
// false->表示cells未初始化，当前所有线程应该将数据写入到base中
//条件二：要执行到条件二，说明条件一是false
// true-> 表示发生竞争了，可能需要重试或者扩容
// false-> 表示当前现成CAS替换数据成功，
if ((as = cells) != null || !casBase(b = base, b + x)) {
//什么时候进入？
//1.条件一 true->表示cells已经初始化过了，当前线程应该将数据写入到对应的cell中
//2.条件二 true-> 表示发生竞争了，可能需要重试或者扩容
// true 表示未发生竞争，false 发生竞争
boolean uncontended = true;
//条件一： true -> cells未初始化，说明此时是通过 2，多线程写base发生竞争进来的
// false -> cells初始化了，说明是那么当前线程应该找自己的cell写值
if (as == null || (m = as.length - 1) < 0 ||
//条件一为false，就走条件二
//条件二：getProbe()：获取当前线程的hash值 getProbe()&m 会 <= m 。因此as[getProbe() & m])就表示当前线程想把数据扔进去的单元格
// true->说明当前线程对应下标的cell为空，需要longAccumulate 创建
// false->说明当前线程对应下标的cell不为空，下一步想要将x值添加到cell中
(a = as[getProbe() & m]) == null ||
//如果条件二位false，就走条件三
//条件三：将x值添加到cell的过程
// true->表示cas添加失败，意味着当前线程对应的cell有竞争
// false->表示cas成功，
!(uncontended = a.cas(v = a.value, v + x)))
//什么时候会调用这个方法？
//1. 条件一： true -> cells未初始化，说明此时是通过 2，多线程写base发生竞争进来的。说明后续需要重试或者初始化cells
//2. 条件二：true->说明当前线程对应下标的cell为空，需要longAccumulate 创建
//3. 条件三：true->表示cas添加失败，意味着当前线程对应的cell有竞争。后续需要重试或者扩容
longAccumulate(x, null, uncontended);
}
}

复制代码

//什么时候会调用这个方法？
//1. 条件一： true -> cells未初始化，说明此时是通过 2，多线程写base发生竞争进来的。说明后续需要重试或者初始化cells
//2. 条件二：true->说明当前线程对应下标的cell为空，需要longAccumulate 创建
//3. 条件三：true->表示cas添加失败，意味着当前线程对应的cell有竞争。后续需要重试或者扩容
//wasUncontended：只有cells初始化之后，并且当前线程竞争修改失败，才会是false
final void longAccumulate(long x, LongBinaryOperator fn,
boolean wasUncontended) {
//当前线程的hash值
int h;
//条件 true->表示当前线程还未分配hash值
if ((h = getProbe()) == 0) {
//因此，执行分配hash值的逻辑
ThreadLocalRandom.current(); // force initialization
h = getProbe();
//为什么？因为在这之前当前线程没有hash值，也就是0，那么当前线程肯定是写入到cells[0]的位置
//如果没有分配hash值的都写到cells[0]，那就出现了cells[0]的竞争。那么就不应该把这次竞争当成真正的竞争，因此修改为true
wasUncontended = true;
}
//表示扩容意向，false：一定不会扩容；true：可能会扩容
boolean collide = false; // True if last slot nonempty
//自旋
for (;;) {
// as 表示cells引用
// a 表示当前线程命中的cell
// n 表示cells数组长度
// v 表示期望值
Cell[] as; Cell a; int n; long v;
//case1：条件1：true -> cells已经初始化，当前线程应该写入数据到对应cell中
// 条件2：true -> 数组长度大于0，与上面一样
if ((as = cells) != null && (n = as.length) > 0) {
//什么时候来到case1
//2. 条件二：true->说明当前线程对应下标的cell为空，需要longAccumulate 创建
//3. 条件三：true->表示cas添加失败，意味着当前线程对应的cell有竞争。后续需要重试或者扩容
//case 1.1：true -> 当前线程对应下标的cell为空，需要创建cell
if ((a = as[(n - 1) & h]) == null) {
//true->当前是无锁未被占用，false->锁被占用
if (cellsBusy == 0) { // Try to attach new Cell
//创建cell
Cell r = new Cell(x); // Optimistically create
//条件一：cellsBusy == 0
// true->当前无锁，当前线程可以竞争这把锁
//条件二：casCellsBusy()，竞争锁
// true->当前线程获取锁
if (cellsBusy == 0 && casCellsBusy()) {
//定义是否创建成功的标记
boolean created = false;
try { // Recheck under lock
//rs表示当前cells引用
//m cells长度
//j 当前线程命中下标
Cell[] rs; int m, j;
//条件一条件二恒成立
//条件三：rs[j = (m - 1) & h] == null？在case1.1时已经判断过这个位置了，为什么这里还要判断？
//原因是多线程并发情况下，有线程可能已经在执行下述流程，此时在case1.1判断为null，但到这里已经有线程执行过了，因此需要重新判断
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
}
}
//扩容意向改为false
collide = false;
}
//case 1.2: wasUncontended：只有cells初始化之后，并且当前线程竞争修改失败，才会是false
else if (!wasUncontended) // CAS already known to fail
wasUncontended = true; // Continue after rehash
//case1.3：什么时候到这，当前线程重置过hash值。新命中的cell不为空
// true->将数据写入cell成功，那就可以退出了
// false->表示重置hash后命中的新的cell也有竞争，重试1次，会执行case1.4
else if (a.cas(v = a.value, ((fn == null) ? v + x :
fn.applyAsLong(v, x))))
break;
//case1.4：n>=NCPU
// true->数组长度大于等于CPU数量
// false->数组长度还可以扩容
// cells != as？
// true->其它线程已经扩容过了，当前线程就应该重置hash重试
else if (n >= NCPU || cells != as)
//扩容意向改为false
collide = false; // At max size or stale
//case1.5：!collide
// true->表示设置扩容意向为true，但不一定扩容，因为需要自旋重新尝试
else if (!collide)
collide = true;
//case 1.6：真正扩容的逻辑
// 条件一：cellsBusy == 0
// true->当前无锁，当前线程可以竞争这把锁
//条件二：casCellsBusy()，竞争锁
// true->当前线程获取锁，当前线程执行可以扩容逻辑
//false就说明有其他线程在执行扩容
else if (cellsBusy == 0 && casCellsBusy()) {
try {
//这里与之前的逻辑一致，需要再次判断
if (cells == as) { // Expand table unless stale
Cell[] rs = new Cell[n << 1];//扩容长度翻倍，长度是2的幂
for (int i = 0; i < n; ++i)
rs[i] = as[i];
cells = rs;
}
} finally {
cellsBusy = 0;
}
collide = false;
continue; // Retry with expanded table
}
//重置当前线程hash值
h = advanceProbe(h);
}
//case2: 显然要到case2，case1就为false,也就是cells还未初始化，as为null
// 条件一：true -> 当前未加锁
// 条件二：cells == as？原因在多线程并发情况下，有线程可能已经在执行下述流程，此时在case1判断为null，但到case2可能已经扩容完成了，cells可能就不为null了
// 条件三：true -> 表示获取锁成功，casCellsBusy() = 1。
// false -> 表示其它线程正在持有锁
else if (cellsBusy == 0 && cells == as && casCellsBusy() = 1。
) {
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;
}
//case3：什么时候会到这个条件。
//1. 当前casCellsBusy()锁已经被持有，说明其他线程正在初始化cells
//2. cells被其他线程初始化了
//那么此时就应该去累加数据了
else if (casBase(v = base, ((fn == null) ? v + x :
fn.applyAsLong(v, x))))
break; // Fall back on using base
}
}

复制代码

public class AtomicStampedReference<V> {
private static class Pair<T> {
final T reference; //维护对象引用
final int stamp; //用于标志版本
private Pair(T reference, int stamp) {
this.reference = reference;
this.stamp = stamp;
}
static <T> Pair<T> of(T reference, int stamp) {
return new Pair<T>(reference, stamp);
}
}
private volatile Pair<V> pair;
....
/**
* expectedReference ：更新之前的原始值
* newReference : 将要更新的新值
* expectedStamp : 期待更新的标志版本
* newStamp : 将要更新的标志版本
*/
public boolean compareAndSet(V expectedReference,
V newReference,
int expectedStamp,
int newStamp) {
// 获取当前的(元素值，版本号)对
Pair<V> current = pair;
return
// 引用没变
expectedReference == current.reference &&
// 版本号没变
expectedStamp == current.stamp &&
// 新引用等于旧引用
((newReference == current.reference &&
// 新版本号等于旧版本号
newStamp == current.stamp) ||
// 构造新的Pair对象并CAS更新
casPair(current, Pair.of(newReference, newStamp)));
}
private boolean casPair(Pair<V> cmp, Pair<V> val) {
// 调用Unsafe的compareAndSwapObject()方法CAS更新pair的引用为新引用
return UNSAFE.compareAndSwapObject(this, pairOffset, cmp, val);
}

复制代码