284 lines
14 KiB
Markdown
284 lines
14 KiB
Markdown
|
# 21 | 原子类：无锁工具类的典范
|
|||
|
|
|
|||
|
|
前面我们多次提到一个累加器的例子，示例代码如下。在这个例子中，add10K()这个方法不是线程安全的，问题就出在变量count的可见性和count+=1的原子性上。可见性问题可以用volatile来解决，而原子性问题我们前面一直都是采用的互斥锁方案。
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
public class Test {
|
|||
|
|
long count = 0;
|
|||
|
|
void add10K() {
|
|||
|
|
int idx = 0;
|
|||
|
|
while(idx++ < 10000) {
|
|||
|
|
count += 1;
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
其实对于简单的原子性问题，还有一种**无锁方案**。Java SDK并发包将这种无锁方案封装提炼之后，实现了一系列的原子类。不过，在深入介绍原子类的实现之前，我们先看看如何利用原子类解决累加器问题，这样你会对原子类有个初步的认识。
|
|||
|
|
|
|||
|
|
在下面的代码中，我们将原来的long型变量count替换为了原子类AtomicLong，原来的 `count +=1` 替换成了 count.getAndIncrement()，仅需要这两处简单的改动就能使add10K()方法变成线程安全的，原子类的使用还是挺简单的。
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
public class Test {
|
|||
|
|
AtomicLong count =
|
|||
|
|
new AtomicLong(0);
|
|||
|
|
void add10K() {
|
|||
|
|
int idx = 0;
|
|||
|
|
while(idx++ < 10000) {
|
|||
|
|
count.getAndIncrement();
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
无锁方案相对互斥锁方案，最大的好处就是**性能**。互斥锁方案为了保证互斥性，需要执行加锁、解锁操作，而加锁、解锁操作本身就消耗性能；同时拿不到锁的线程还会进入阻塞状态，进而触发线程切换，线程切换对性能的消耗也很大。 相比之下，无锁方案则完全没有加锁、解锁的性能消耗，同时还能保证互斥性，既解决了问题，又没有带来新的问题，可谓绝佳方案。那它是如何做到的呢？
|
|||
|
|
|
|||
|
|
## 无锁方案的实现原理
|
|||
|
|
|
|||
|
|
其实原子类性能高的秘密很简单，硬件支持而已。CPU为了解决并发问题，提供了CAS指令（CAS，全称是Compare And Swap，即“比较并交换”）。CAS指令包含3个参数：共享变量的内存地址A、用于比较的值B和共享变量的新值C；并且只有当内存中地址A处的值等于B时，才能将内存中地址A处的值更新为新值C。**作为一条CPU指令，CAS指令本身是能够保证原子性的**。
|
|||
|
|
|
|||
|
|
你可以通过下面CAS指令的模拟代码来理解CAS的工作原理。在下面的模拟程序中有两个参数，一个是期望值expect，另一个是需要写入的新值newValue，**只有当目前count的值和期望值expect相等时，才会将count更新为newValue**。
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
class SimulatedCAS{
|
|||
|
|
int count；
|
|||
|
|
synchronized int cas(
|
|||
|
|
int expect, int newValue){
|
|||
|
|
// 读目前count的值
|
|||
|
|
int curValue = count;
|
|||
|
|
// 比较目前count值是否==期望值
|
|||
|
|
if(curValue == expect){
|
|||
|
|
// 如果是，则更新count的值
|
|||
|
|
count = newValue;
|
|||
|
|
}
|
|||
|
|
// 返回写入前的值
|
|||
|
|
return curValue;
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
你仔细地再次思考一下这句话，“**只有当目前count的值和期望值expect相等时，才会将count更新为newValue。**”要怎么理解这句话呢？
|
|||
|
|
|
|||
|
|
对于前面提到的累加器的例子，`count += 1` 的一个核心问题是：基于内存中count的当前值A计算出来的count+=1为A+1，在将A+1写入内存的时候，很可能此时内存中count已经被其他线程更新过了，这样就会导致错误地覆盖其他线程写入的值（如果你觉得理解起来还有困难，建议你再重新看看[《01 | 可见性、原子性和有序性问题：并发编程Bug的源头》](https://time.geekbang.org/column/article/83682)）。也就是说，只有当内存中count的值等于期望值A时，才能将内存中count的值更新为计算结果A+1，这不就是CAS的语义吗！
|
|||
|
|
|
|||
|
|
使用CAS来解决并发问题，一般都会伴随着自旋，而所谓自旋，其实就是循环尝试。例如，实现一个线程安全的`count += 1`操作，“CAS+自旋”的实现方案如下所示，首先计算newValue = count+1，如果cas(count,newValue)返回的值不等于count，则意味着线程在执行完代码①处之后，执行代码②处之前，count的值被其他线程更新过。那此时该怎么处理呢？可以采用自旋方案，就像下面代码中展示的，可以重新读count最新的值来计算newValue并尝试再次更新，直到成功。
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
class SimulatedCAS{
|
|||
|
|
volatile int count;
|
|||
|
|
// 实现count+=1
|
|||
|
|
addOne(){
|
|||
|
|
do {
|
|||
|
|
newValue = count+1; //①
|
|||
|
|
}while(count !=
|
|||
|
|
cas(count,newValue) //②
|
|||
|
|
}
|
|||
|
|
// 模拟实现CAS，仅用来帮助理解
|
|||
|
|
synchronized int cas(
|
|||
|
|
int expect, int newValue){
|
|||
|
|
// 读目前count的值
|
|||
|
|
int curValue = count;
|
|||
|
|
// 比较目前count值是否==期望值
|
|||
|
|
if(curValue == expect){
|
|||
|
|
// 如果是，则更新count的值
|
|||
|
|
count= newValue;
|
|||
|
|
}
|
|||
|
|
// 返回写入前的值
|
|||
|
|
return curValue;
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
通过上面的示例代码，想必你已经发现了，CAS这种无锁方案，完全没有加锁、解锁操作，即便两个线程完全同时执行addOne()方法，也不会有线程被阻塞，所以相对于互斥锁方案来说，性能好了很多。
|
|||
|
|
|
|||
|
|
但是在CAS方案中，有一个问题可能会常被你忽略，那就是**ABA**的问题。什么是ABA问题呢？
|
|||
|
|
|
|||
|
|
前面我们提到“如果cas(count,newValue)返回的值**不等于**count，意味着线程在执行完代码①处之后，执行代码②处之前，count的值被其他线程**更新过**”，那如果cas(count,newValue)返回的值**等于**count，是否就能够认为count的值没有被其他线程**更新过**呢？显然不是的，假设count原本是A，线程T1在执行完代码①处之后，执行代码②处之前，有可能count被线程T2更新成了B，之后又被T3更新回了A，这样线程T1虽然看到的一直是A，但是其实已经被其他线程更新过了，这就是ABA问题。
|
|||
|
|
|
|||
|
|
可能大多数情况下我们并不关心ABA问题，例如数值的原子递增，但也不能所有情况下都不关心，例如原子化的更新对象很可能就需要关心ABA问题，因为两个A虽然相等，但是第二个A的属性可能已经发生变化了。所以在使用CAS方案的时候，一定要先check一下。
|
|||
|
|
|
|||
|
|
## 看Java如何实现原子化的count += 1
|
|||
|
|
|
|||
|
|
在本文开始部分，我们使用原子类AtomicLong的getAndIncrement()方法替代了`count += 1`，从而实现了线程安全。原子类AtomicLong的getAndIncrement()方法内部就是基于CAS实现的，下面我们来看看Java是如何使用CAS来实现原子化的`count += 1`的。
|
|||
|
|
|
|||
|
|
在Java 1.8版本中，getAndIncrement()方法会转调unsafe.getAndAddLong()方法。这里this和valueOffset两个参数可以唯一确定共享变量的内存地址。
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
final long getAndIncrement() {
|
|||
|
|
return unsafe.getAndAddLong(
|
|||
|
|
this, valueOffset, 1L);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
unsafe.getAndAddLong()方法的源码如下，该方法首先会在内存中读取共享变量的值，之后循环调用compareAndSwapLong()方法来尝试设置共享变量的值，直到成功为止。compareAndSwapLong()是一个native方法，只有当内存中共享变量的值等于expected时，才会将共享变量的值更新为x，并且返回true；否则返回fasle。compareAndSwapLong的语义和CAS指令的语义的差别仅仅是返回值不同而已。
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
public final long getAndAddLong(
|
|||
|
|
Object o, long offset, long delta){
|
|||
|
|
long v;
|
|||
|
|
do {
|
|||
|
|
// 读取内存中的值
|
|||
|
|
v = getLongVolatile(o, offset);
|
|||
|
|
} while (!compareAndSwapLong(
|
|||
|
|
o, offset, v, v + delta));
|
|||
|
|
return v;
|
|||
|
|
}
|
|||
|
|
//原子性地将变量更新为x
|
|||
|
|
//条件是内存中的值等于expected
|
|||
|
|
//更新成功则返回true
|
|||
|
|
native boolean compareAndSwapLong(
|
|||
|
|
Object o, long offset,
|
|||
|
|
long expected,
|
|||
|
|
long x);
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
另外，需要你注意的是，getAndAddLong()方法的实现，基本上就是CAS使用的经典范例。所以请你再次体会下面这段抽象后的代码片段，它在很多无锁程序中经常出现。Java提供的原子类里面CAS一般被实现为compareAndSet()，compareAndSet()的语义和CAS指令的语义的差别仅仅是返回值不同而已，compareAndSet()里面如果更新成功，则会返回true，否则返回false。
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
do {
|
|||
|
|
// 获取当前值
|
|||
|
|
oldV = xxxx；
|
|||
|
|
// 根据当前值计算新值
|
|||
|
|
newV = ...oldV...
|
|||
|
|
}while(!compareAndSet(oldV,newV);
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
## 原子类概览
|
|||
|
|
|
|||
|
|
Java SDK并发包里提供的原子类内容很丰富，我们可以将它们分为五个类别：**原子化的基本数据类型、原子化的对象引用类型、原子化数组、原子化对象属性更新器**和**原子化的累加器**。这五个类别提供的方法基本上是相似的，并且每个类别都有若干原子类，你可以通过下面的原子类组成概览图来获得一个全局的印象。下面我们详细解读这五个类别。
|
|||
|
|
|
|||
|
|
|
|||
|
|
|
|||
|
|
原子类组成概览图
|
|||
|
|
|
|||
|
|
### 1\. 原子化的基本数据类型
|
|||
|
|
|
|||
|
|
相关实现有AtomicBoolean、AtomicInteger和AtomicLong，提供的方法主要有以下这些，详情你可以参考SDK的源代码，都很简单，这里就不详细介绍了。
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
getAndIncrement() //原子化i++
|
|||
|
|
getAndDecrement() //原子化的i--
|
|||
|
|
incrementAndGet() //原子化的++i
|
|||
|
|
decrementAndGet() //原子化的--i
|
|||
|
|
//当前值+=delta，返回+=前的值
|
|||
|
|
getAndAdd(delta)
|
|||
|
|
//当前值+=delta，返回+=后的值
|
|||
|
|
addAndGet(delta)
|
|||
|
|
//CAS操作，返回是否成功
|
|||
|
|
compareAndSet(expect, update)
|
|||
|
|
//以下四个方法
|
|||
|
|
//新值可以通过传入func函数来计算
|
|||
|
|
getAndUpdate(func)
|
|||
|
|
updateAndGet(func)
|
|||
|
|
getAndAccumulate(x,func)
|
|||
|
|
accumulateAndGet(x,func)
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
### 2\. 原子化的对象引用类型
|
|||
|
|
|
|||
|
|
相关实现有AtomicReference、AtomicStampedReference和AtomicMarkableReference，利用它们可以实现对象引用的原子化更新。AtomicReference提供的方法和原子化的基本数据类型差不多，这里不再赘述。不过需要注意的是，对象引用的更新需要重点关注ABA问题，AtomicStampedReference和AtomicMarkableReference这两个原子类可以解决ABA问题。
|
|||
|
|
|
|||
|
|
解决ABA问题的思路其实很简单，增加一个版本号维度就可以了，这个和我们在[《18 | StampedLock：有没有比读写锁更快的锁？》](https://time.geekbang.org/column/article/89456)介绍的乐观锁机制很类似，每次执行CAS操作，附加再更新一个版本号，只要保证版本号是递增的，那么即便A变成B之后再变回A，版本号也不会变回来（版本号递增的）。AtomicStampedReference实现的CAS方法就增加了版本号参数，方法签名如下：
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
boolean compareAndSet(
|
|||
|
|
V expectedReference,
|
|||
|
|
V newReference,
|
|||
|
|
int expectedStamp,
|
|||
|
|
int newStamp)
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
AtomicMarkableReference的实现机制则更简单，将版本号简化成了一个Boolean值，方法签名如下：
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
boolean compareAndSet(
|
|||
|
|
V expectedReference,
|
|||
|
|
V newReference,
|
|||
|
|
boolean expectedMark,
|
|||
|
|
boolean newMark)
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
### 3\. 原子化数组
|
|||
|
|
|
|||
|
|
相关实现有AtomicIntegerArray、AtomicLongArray和AtomicReferenceArray，利用这些原子类，我们可以原子化地更新数组里面的每一个元素。这些类提供的方法和原子化的基本数据类型的区别仅仅是：每个方法多了一个数组的索引参数，所以这里也不再赘述了。
|
|||
|
|
|
|||
|
|
### 4\. 原子化对象属性更新器
|
|||
|
|
|
|||
|
|
相关实现有AtomicIntegerFieldUpdater、AtomicLongFieldUpdater和AtomicReferenceFieldUpdater，利用它们可以原子化地更新对象的属性，这三个方法都是利用反射机制实现的，创建更新器的方法如下：
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
public static <U>
|
|||
|
|
AtomicXXXFieldUpdater<U>
|
|||
|
|
newUpdater(Class<U> tclass,
|
|||
|
|
String fieldName)
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
需要注意的是，**对象属性必须是volatile类型的，只有这样才能保证可见性**；如果对象属性不是volatile类型的，newUpdater()方法会抛出IllegalArgumentException这个运行时异常。
|
|||
|
|
|
|||
|
|
你会发现newUpdater()的方法参数只有类的信息，没有对象的引用，而更新**对象**的属性，一定需要对象的引用，那这个参数是在哪里传入的呢？是在原子操作的方法参数中传入的。例如compareAndSet()这个原子操作，相比原子化的基本数据类型多了一个对象引用obj。原子化对象属性更新器相关的方法，相比原子化的基本数据类型仅仅是多了对象引用参数，所以这里也不再赘述了。
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
boolean compareAndSet(
|
|||
|
|
T obj,
|
|||
|
|
int expect,
|
|||
|
|
int update)
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
### 5\. 原子化的累加器
|
|||
|
|
|
|||
|
|
DoubleAccumulator、DoubleAdder、LongAccumulator和LongAdder，这四个类仅仅用来执行累加操作，相比原子化的基本数据类型，速度更快，但是不支持compareAndSet()方法。如果你仅仅需要累加操作，使用原子化的累加器性能会更好。
|
|||
|
|
|
|||
|
|
## 总结
|
|||
|
|
|
|||
|
|
无锁方案相对于互斥锁方案，优点非常多，首先性能好，其次是基本不会出现死锁问题（但可能出现饥饿和活锁问题，因为自旋会反复重试）。Java提供的原子类大部分都实现了compareAndSet()方法，基于compareAndSet()方法，你可以构建自己的无锁数据结构，但是**建议你不要这样做，这个工作最好还是让大师们去完成**，原因是无锁算法没你想象的那么简单。
|
|||
|
|
|
|||
|
|
Java提供的原子类能够解决一些简单的原子性问题，但你可能会发现，上面我们所有原子类的方法都是针对一个共享变量的，如果你需要解决多个变量的原子性问题，建议还是使用互斥锁方案。原子类虽好，但使用要慎之又慎。
|
|||
|
|
|
|||
|
|
## 课后思考
|
|||
|
|
|
|||
|
|
下面的示例代码是合理库存的原子化实现，仅实现了设置库存上限setUpper()方法，你觉得setUpper()方法的实现是否正确呢？
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
public class SafeWM {
|
|||
|
|
class WMRange{
|
|||
|
|
final int upper;
|
|||
|
|
final int lower;
|
|||
|
|
WMRange(int upper,int lower){
|
|||
|
|
//省略构造函数实现
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
final AtomicReference<WMRange>
|
|||
|
|
rf = new AtomicReference<>(
|
|||
|
|
new WMRange(0,0)
|
|||
|
|
);
|
|||
|
|
// 设置库存上限
|
|||
|
|
void setUpper(int v){
|
|||
|
|
WMRange nr;
|
|||
|
|
WMRange or = rf.get();
|
|||
|
|
do{
|
|||
|
|
// 检查参数合法性
|
|||
|
|
if(v < or.lower){
|
|||
|
|
throw new IllegalArgumentException();
|
|||
|
|
}
|
|||
|
|
nr = new
|
|||
|
|
WMRange(v, or.lower);
|
|||
|
|
}while(!rf.compareAndSet(or, nr));
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
```
|
|||
|
|
|
|||
|
|
欢迎在留言区与我分享你的想法，也欢迎你在留言区记录你的思考过程。感谢阅读，如果你觉得这篇文章对你有帮助的话，也欢迎把它分享给更多的朋友。
|
|||
|
|
|