Java多线程-CAS比较交换

发表于 | 分类于 ，

CAS

  • CAS：compare and swap，比较交换
  • CAS有三个参数CAS(V,E,N)
    • V表示要更新的变量
    • E表示预期值
    • N表示新值
    • 只有V=E相等时，才会将V更新为N,否则说明其他线程进行了更新，什么也不做。
    • 失败会被告知，允许再次尝试
1
2
3
4
5
6
7
8
public int a = 1;
public boolean compareAndSwapInt(int b) {
    if (a == 1) {
        a = b;
        return true;
    }
    return false;
}
  • 上面的代码中，多个线程同时更改a的值时，无法确定a最后的值，对compareAndSwapInt进行加锁，使其成为一个原子操作，同一个时刻只有一个线程才可以改变a的值。
  • CAS中的比较和替换是一组原子操作，不会被外部打断，先根据paramLong回去到当前内存中的值V，然后将内存值V和原值E做比较，相等则将V替换为N,效率要高于加锁的操作。

AtomicInteger

  • 无锁的线程安全整数类

成员变量

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public class AtomicInteger extends Number implements java.io.Serializable {
    private static final long serialVersionUID = 6214790243416807050L;
    // java通过native来访问底层操作系统，Unsafe类提供了硬件级别的原子操作
    private static final Unsafe unsafe = Unsafe.getUnsafe();
    private static final long valueOffset;
    static {
        try {
            // 变量值在内存中的偏移地址
            valueOffset = unsafe.objectFieldOffset
                (AtomicInteger.class.getDeclaredField("value"));
        } catch (Exception ex) { throw new Error(ex); }
    }
    // value使用volatile修饰，从而多个线程之间可见
    private volatile int value;
    // ...
}

构造方法

1
2
3
4
5
6
7
public AtomicInteger(int initialValue) {
    value = initialValue;
}
public AtomicInteger() {
}

普通方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// 读
public final int get() {
    return value;
}
// 写
public final void set(int newValue) {
    value = newValue;
}
// 延迟写
public final void lazySet(int newValue) {
    unsafe.putOrderedInt(this, valueOffset, newValue);
}

原子方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
/**
  原子操作，更新值，返回旧值
 */
public final int getAndSet(int newValue) {
    return unsafe.getAndSetInt(this, valueOffset, newValue);
}
public final int getAndSetInt(Object var1, long var2, int var4) {
    int var5;
    do {
        var5 = this.getIntVolatile(var1, var2);
    } while(!this.compareAndSwapInt(var1, var2, var5, var4));
    return var5;
}
// 更新值，成功返回true
public final boolean compareAndSet(int expect, int update) {
    return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
}
// 原子累加
public final int getAndIncrement() {
    return unsafe.getAndAddInt(this, valueOffset, 1);
}
// 原子递减
public final int getAndDecrement() {
    return unsafe.getAndAddInt(this, valueOffset, -1);
}
// 原子增加
public final int getAndAdd(int delta) {
    return unsafe.getAndAddInt(this, valueOffset, delta);
}

实例

  • AtomicInteger累加
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
package third;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
public class AtomicIntegerDemo {
    private static AtomicInteger integer = new AtomicInteger();
    public static void main(String[] args) throws InterruptedException {
        ExecutorService service = Executors.newFixedThreadPool(10);
        CountDownLatch cd = new CountDownLatch(10);
        for (int i = 0; i < 10; i++) {
            service.execute(new AddThread(cd, integer));
        }
        cd.await();
        System.out.println(integer);
    }
}
class AddThread implements Runnable {
    private CountDownLatch cd;
    private AtomicInteger integer;
    public AddThread(CountDownLatch cd, AtomicInteger integer) {
        this.cd = cd;
        this.integer = integer;
    }
    @Override
    public void run() {
        for (int k = 0; k < 10000; k++)
            integer.incrementAndGet();
        cd.countDown();
    }
}
// 100000
  • AtomicIntegerArray累加
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
public class AtomicIntegerArrayDemo {
    public static void main(String[] args) throws InterruptedException {
        AtomicIntegerArray integers = new AtomicIntegerArray(10);
        CountDownLatch latch = new CountDownLatch(10);
        ExecutorService service = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 10; i++)
            service.execute(new AddArrThread(integers, latch));
        latch.await();
        System.out.println(integers);
    }
}
class AddArrThread implements Runnable {
    private AtomicIntegerArray integers;
    private CountDownLatch cd;
    public AddArrThread(AtomicIntegerArray integers, CountDownLatch cd) {
        this.integers = integers;
        this.cd = cd;
    }
    @Override
    public void run() {
        for (int k = 0; k < 10000; k++)
            integers.getAndIncrement(k % (integers.length()-2));
        cd.countDown();
    }
}
Donate comment here