乐观锁:高效并发无锁方案
4.乐观锁
这一章主要介绍乐观锁。前面的管程部分讲了悲观锁,现在做一些总结:
- 悲观锁(Pessimistic Lock):悲观锁认为数据在多线程或多进程环境下总是容易发生冲突/冲突的概率高,所以在数据操作前,直接加锁(独占),别的线程/进程不能再访问或只能阻塞等待。因此悲观锁实用与写操作多、共享资源竞争激烈以及一致性要求高的场景。但是如果面对读操作多的情况,使用悲观锁会导致线程切换进而导致额外开销,就不如乐观锁。
- 数据库层 :
SELECT ... FOR UPDATE
,行锁、表锁 - Java并发 :
synchronized
,ReentrantLock
,AQS独占锁等 - 其他 :文件锁、操作系统互斥锁
- 乐观锁(Optimistic Lock):乐观锁假定冲突发生的概率低,不加锁,先操作,最后再校验 (即使有冲突,再补救)。因此乐观锁的过程是不涉及真实的锁,而是使用有冲突再补救的思想进行判断,因此不会涉及阻塞和互斥,性能较好,适用于读多写少的场景。但问题是如果在写多的场景下,会经常出现重试,也就会导致效率降低。且如果对于乐观锁的粒度要求高,会出现ABA问题。
- CAS(Compare and Swap)机制 —— Java原子类
AtomicInteger
、AtomicReference
等 - 版本号机制 — 每条数据/对象带版本号,更新时校对版本号
- 数据库乐观锁 :如在UPDATE/DELETE时加条件“WHERE version = ?”,更新不成代表有别的事务改过了,重试
4.1 CAS
如果不加synchronized则会出现线程安全的问题,即cost不会为0。加synchronized则会解决线程安全的问题。即会因为多个线程操作amount可能会出现并发的问题。例如减余额的操作分为读取值、减余额,写回,这样两个线程都执行前两步,结果只扣了一次。
import java.util.ArrayList;
import java.util.List;
public class Main {public static void main(String[] args) {Account account = new AccountUnsafe(10000);Account.demo(account);}
}class AccountUnsafe implements Account {private Integer balance;public AccountUnsafe(Integer balance) {this.balance = balance;}@Overridepublic Integer getBalance() {synchronized (this) {return this.balance;}}@Overridepublic void withdraw(Integer amount) {this.balance -= amount;}
}interface Account{// 获取余额Integer getBalance();// 取款void withdraw(Integer amount);/*** 方法内会启动1000个线程,每个线程做-10元的操作* 如果初始余额为10000那么正确的结果应当是0*/static void demo(Account account) {List<Thread> ts = new ArrayList<>();for (int i = 0; i < 1000; i++) {ts.add(new Thread(() -> {account.withdraw(10);}));}long start = System.nanoTime();ts.forEach(Thread::start);ts.forEach(t -> {try {t.join();} catch (InterruptedException e) {e.printStackTrace();}});long end = System.nanoTime();System.out.println(account.getBalance()+ " cost: " + (end - start) / 1000_000 + " ms");}
}
// 无锁实现 也可以做到全售卖出 整个过程不涉及sync
class AccountCas implements Account {private AtomicInteger balance;public AccountCas(int balance) {this.balance = new AtomicInteger(balance);}@Overridepublic Integer getBalance() {return balance.get();}@Overridepublic void withdraw(Integer amount) {while(true) {// 获取余额的最新值int prev = balance.get();// 要修改的余额int next = prev - amount;// 真正修改if(balance.compareAndSet(prev, next)) {break;}}}}
无锁原理:
重点在于compareAndSet,即CAS。比较的双方如下:
prev
:调用balance.get()
时读取的余额值(旧值)。balance
的当前实际值:执行CAS时内存中的最新值。
也就是,检查balance
的当前值是否等于prev
(即判断是否有其他线程在此期间修改了余额)。如果是等于,则说明并未被修改,因此将next赋值给balance,并且立即写入主内存。反之则继续循环。
而怎么保证cas操作是原子的?现代处理器提供了专门的原子指令来实现CAS操作,即只有一条指令来进行CAS。
CAS和volatile:
我们使用的AtomicInteger,具体值是使用volatile进行修饰,因此会插入内存屏障保证其可见性。如果不加入volatile,则可能一直看不到最新值,导致一直死循环,因此CAS必须配合volatile才可以使用。
无锁效率:
无锁的效率更高,因为cas的自选操作不需要切换上下文,而sync需要切换上下文,多出开销。但是CAS需要额外的cpu的支持,不然就没有实际意义。也因此最好线程数不要超过cpu数,超过了再自旋也没有意义。
总结:
- cas是乐观锁的思想,即认为发生冲突的情况很少,遇到冲突则会重试。而sync是悲观锁,假设冲突经常发生。
- cas是无锁且无阻塞的并发。即真的不加锁,且状态不改变不是阻塞态。如果竞争如果十分激烈,重试经常发生则会效率降低。
4.2 无锁线程安全工具,原子类
原子类的全部方法都保证了原子性,其原子性的原因仍为cas+volatile。例如AtomicInteger的源码有一个int value代表值,这个变量就是使用volatile修饰,每次读取一定读取到最新值,不会脏读。CAS则是指令保证的。
这里说一下,Integer、String等类都是线程安全的,也就是不可变类,即一旦创建状态就不可以修改。也因此多线程的读写不会影响对象本身。但下面的操作就发生了读-写-改的操作,多线程下这样的操作就不是原子的,因此原子类就是解决类似读-写-改这样的多次操作,从而保证多线程下的原子性。
Integer a = 1;
// Thread A
a = a + 1; // 等价于 int tmp = a.intValue(); tmp++; a = new Integer(tmp);
// Thread B
a = a + 1;
4.2.1 原子整数
有如下方法,方法都保证原子性:
AtomicInteger i = new AtomicInteger(0);// 获取并自增(i = 0, 结果 i = 1, 返回 0),类似于 i++
System.out.println(i.getAndIncrement());// 自增并获取(i = 1, 结果 i = 2, 返回 2),类似于 ++i
System.out.println(i.incrementAndGet());// 自减并获取(i = 2, 结果 i = 1, 返回 1),类似于 --i
System.out.println(i.decrementAndGet());// 获取并自减(i = 1, 结果 i = 0, 返回 1),类似于 i--
System.out.println(i.getAndDecrement());// 获取并加值(i = 0, 结果 i = 5, 返回 0)
System.out.println(i.getAndAdd(5));// 加值并获取(i = 5, 结果 i = 10, 返回 10)
System.out.println(i.addAndGet(5));// 更新并获取,适合复杂操作。即将得到的value值*10再获取,输出为100.
System.out.println(i.updateAndGet(value -> value*10));
因此有了上面的原子的方法,我们之前的代码就可以做如下修改。更改后的代码不需要在手动写CAS进行循环判断,而是交给getAndAdd方法,其内部也是CAS的操作。
@Override
public void withdraw(Integer amount) {/*while(true) {// 获取余额的最新值int prev = balance.get();// 要修改的余额int next = prev - amount;// 真正修改if(balance.compareAndSet(prev, next)) {break;}}*/balance.getAndAdd(delta: -1*amount);
}
updateAndGet原理:
思想还是使用compareAndSet进行判断。不过需要传入一个一元的操作符来代表操作。
public final int updateAndGet(IntUnaryOperator updateFunction) {int prev, next;do {prev = get();next = updateFunction.applyAsInt(prev);} while (!compareAndSet(prev, next));return next;
}
4.2.2 原子引用
想保护的不一定都是基本数据类型,因此使用原子引用类型。比如我们把上面的例子的Integer类型替换为BigDecimal类型,那就不能使用AtomicInteger类型,可以使用原子引用类型。
class DecimalAccountCas implements DecimalAccount {private AtomicReference<BigDecimal> balance; // 在泛型的部分加入需要替换的类型public DecimalAccountCas(BigDecimal balance) { // 构造函数,传入balancethis.balance = new AtomicReference<>(balance);}@Overridepublic BigDecimal getBalance() {return balance.get();}@Overridepublic void withdraw(BigDecimal amount) {while(true) {BigDecimal prev = balance.get();BigDecimal next = prev.subtract(amount);if (balance.compareAndSet(prev, next)) {break;}}}
}
原子引用的ABA问题:
例如有如下一段代码:
public class Test {static AtomicReference<String> ref = new AtomicReference<>("A");public static void main(String[] args) throws InterruptedException {Log.debug("main start...");// 获取值 AString prev = ref.get();other();sleep(1);// 尝试改为 CLog.debug("change A->C {}", ref.compareAndSet(prev, update: "C"));}private static void other() {new Thread(() -> {Log.debug("change A->B {}", ref.compareAndSet(ref.get(), update: "B"));}, name: "t1").start();sleep(0.5);new Thread(() -> {Log.debug("change B->A {}", ref.compareAndSet(ref.get(), update: "A"));}, name: "t2").start();}
}
这一段代码的main线程会sleep 1s,期间other会主动改动ref的值为B再改为A,但是main线程无法判断是否有其他线程更改过。因此,CAS无法判断是否有其他线程进行修改。虽然大部分情况这种问题不会导致真正的问题,因为最后main判断的数据仍然为A,但是如果希望只要发生修改就返回false,则仅仅比较值是不够的,还需要增加一个版本号。每次修改都会更改版本号。需要替换原子引用,使用AtomicStampedReference,即原子版本号引用。
如下代码,other中的两个线程都可以修改,并且会递增版本号,但是main线程则会由于版本号对不上CAS失败。
public class AtomicStampedReferenceExample {static AtomicStampedReference<String> ref = new AtomicStampedReference<>("A", 0);public static void main(String[] args) throws InterruptedException {Log.debug("main start...");// 获取值 AString prev = ref.getReference();int stamp = ref.getStamp();other();// 尝试改为 CLog.debug("change A->C {}", ref.compareAndSet(prev, "C", stamp, stamp + 1));}private static void other() {new Thread(() -> {int stamp = ref.getStamp();Log.debug("change A->B {}", ref.compareAndSet(ref.getReference(), "B", stamp, stamp + 1));}, "t1").start();sleep(0.5);new Thread(() -> {int stamp = ref.getStamp();Log.debug("change B->A {}", ref.compareAndSet(ref.getReference(), "A", stamp, stamp + 1));}, "t2").start();}
}
有时候,不关心更改多少次而是是否做过更改,因此就可以使用AtomicMarkableReference。
ABA问题主要是在乐观锁/CAS并发更新时,内存(或数据/引用)先变成A→B,然后又变回A,因此会误以为没有被修改,实则被修改了。主要出现在原子引用/CAS操作时出现。
4.2.3 原子数组
是一种原子的,可以在多线程下使用的数组。其实就是可以替换数组,也就是int[]
private static <T> void demo(Supplier<T[]> arraySupplier,Function<T[], Integer> lengthFun,BiConsumer<T[], Integer> putConsumer,Consumer<T[]> printConsumer) {List<Thread> ts = new ArrayList<>();T[] array = arraySupplier.get();int length = lengthFun.apply(array);for (int i = 0; i < length; i++) {// 每个线程对数组作10000次操作ts.add(new Thread(() -> {for (int j = 0; j < 10000; j++) {putConsumer.accept(array, j % length);}}));}
}
// 测试函数
demo(// 换成Atomic就不会出错// 如果是() -> new int[10] 就会因为线程不安全导致出错。() -> new AtomicIntegerArray(length: 10),(array) -> array.length(),(array, index) -> array.getAndIncrement(index),array -> System.out.println(array)
);
4.2.4字段更新器:
字段更新器,是更新一个类的某个字段的方式,这个字段必须要被volatile进行修饰。例如一个class里有一个叫account的属性,那么可以通过字段更新器进行原子更改(使用原子方法,例如cas,getAndAnd)。相比于直接使用原子整数或者原子引用,字段更新器不会额外创建原子对象,因此开销更小。但是属性一定要被volatile修饰,且不能为static。
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;public class FieldUpdaterExample {static class BankAccount {private volatile int balance; // 必须是 volatileprivate static final AtomicIntegerFieldUpdater<BankAccount> updater =AtomicIntegerFieldUpdater.newUpdater(BankAccount.class, "balance"); // 对BankAccount的balance字段进行更改。public void withdraw(int amount) {updater.addAndGet(this, -amount); // 原子扣减}public int getBalance() {return balance;}}public static void main(String[] args) {BankAccount account = new BankAccount();account.withdraw(100); // 原子操作System.out.println("Balance: " + account.getBalance());}
}
对于除了Integer的对象,也有对应类型的字段更新器:
AtomicIntegerFieldUpdater //用于 int 类型字段。
AtomicLongFieldUpdater //用于 long 类型字段。
AtomicReferenceFieldUpdater //用于引用类型字段(如 String、自定义对象)。
4.2.5 原子累加器
顾名思义,就是做累加操作的。
4.3 LongAdder
用于在高并发的场景下替代AtomicLong,因为AtomicLong是基于CAS的,如果在高并发下多线程进行竞争可能会导致性能瓶颈。而LongAdder将计数分散到多个变量(多个cell),不同的线程竞争不同的cell,进而减少竞争。
有如下关键字段:
transient volatile Cell[] cells;transient volatile long base;transient volatile int cellsBusy; // 表示cells初始化/扩容等是否在进行中
base
:计数基础值。并发不高时,线程直接尝试 CAS 操作它。cells
:Cell 数组,高并发时,线程 hash 到不同 Cell 进行累加。cellsBusy
:自旋锁,保证 cells 操作线程安全。
而这里的cell本质是一个包含 volatile long 的对象,并能支持自身的 CAS 操作。
// 防止缓存行的伪共享
@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);}
}
这里的注解是防止伪共享。什么是伪共享呢?伪共享 是多线程并发编程中的一种性能隐患。当多个线程操作不同变量 ,但这些变量共享同一个CPU缓存行时,如果线程频繁写操作这些变量,会导致缓存行反复失效(cache line flush),极大降低并发性能。这种情况下虽然线程没有真正访问同一份数据,但会“间接竞争”同一缓存行。
cpu是分级缓存的,我们这里只假设有一个cache,对于多核cpu来说,每个cpu则会配置一个cache。但是内存只有一份。一个cache行和内存行的大小是相同的,都是64字节,而存储的cell则是24字节(对象头16+一个long变量8),**也就是说一个内存行可以存储两个cell。**如果一个cpu1将cell0的值+1,而另一个cpu2只是将cell1+1,那么对应到内存中会触发同一cache line的竞争和cache一致性协议(如MESI)下的共享失效(false sharing),因此会导致一致性问题。
解决方法也就是加入注解:Cell 填充(padding)——@sun.misc.Contended
@sun.misc.Contended
static final class Cell {volatile long value;// ...
}
JVM看到 @Contended
,会自动帮Cell对象在value
前后填充一些“无用字节”(padding),防止同一个Cell数组里的Cell落在同一缓存行,提高并发写性能。简单来说,也就是每个cell独占一个cache行,不会出现多个cell挤在一个cache行的情况。
常用方法:
- add(long x):
增加指定的值(x可以是正数或负数,即加也可以减)。 - increment():
自增1,等价于add(1)
。 - decrement():
自减1,等价于add(-1)
。 - sum():
获取当前所有Cell+base的累计和(近似实时值,不是强一致性)。这里是即使快照,可能在取得sum的时候有其他线程进行修改,但只是获取的值可能是上一个版本的值,但是能保证值是正确的。 - sumThenReset():
获取累计和,并顺带清空为0(适合做周期性采集和清空,比如监控流量时段统计)。 - reset():
直接重置所有统计为0,但不返回和。
源码如下: 这里的cells就是Longadder的cells数组,而cells是懒创建的,没有竞争时就不创建cells。base也是Longadder的base字段,如果base的CAS失败,则说明出现竞争。这两点都说明竞争出现了,进入if分支。如果没有发生竞争,则在casBase的地方就直接进行了累加。
第二个if分支则是查看是否要进入复杂分支,即简单路径是否失败。
public void add(long x) {Cell[] as; long b, v;int m;Cell a;// 1. 如果cells数组已初始化 或 base的CAS失败 if ((as = cells) != null || !casBase(b = base, b + x)) {boolean uncontended = true;// 2. 尝试定位到cell if (as == null || (m = as.length - 1) < 0 ||(a = as[getProbe() & m]) == null ||!(uncontended = a.cas(v = a.value, v + x)))// 3. 碰撞等特殊情况,进入更复杂的处理longAccumulate(x, null, uncontended);}
}
4.4 Unsafe
Unsafe
是 sun.misc.Unsafe
类的简称,是JVM提供的一个底层操作工具类。它允许你直接操纵内存、对象、线程、CAS原语等,抛开了 Java 普通语法和安全机制。因此这里的unsafe是指编程人员直接操作很不安全,而不是这个类不安全。