《读书笔记》程序员的自我修养之线程安全问题

场景：由于多线程程序处于一个多变的环境中，可访问的全局变量和堆数据随时都可能被其他线程改变。

一个经典实例来阐述多个线程同时访问一个共享数据所造成的后果。

线程1	线程2
i=1； ++i；	--i；

首先要明白++i的实现步骤如下：

（1） 读取i到某个寄存器X；

（2） X++；

（3） 将X的内容存储回i。

如果线程1和线程2并发执行，则执行顺序如下：

执行序号	执行指令	语句执行后变量值	线程
1	i=1	i=1，X[1]=未知	1
2	X[1]=i	i=1，X[1]=1	1
3	X[2]=i	i=1，X[2]=1	2
4	X[1]++	i=1，X[1]=2	1
5	X[2]--	i=1，X[2]=0	2
6	i=X[1]	i=2，X[1]=2	1
7	i=X[2]	i=0，X[2]=0	2

从程序逻辑来看，两个线程都执行完之后，i的值应该为1，但是实际的情况下，可能为0，1，2；

造成这个问题的原因是++i这条语句会被编译为3条汇编代码。

可见，两个程序同时读写同一个共享数据时会导致意想不到的后果。

为了避免上述问题，有如下几种方式：

原子指令：单指令操作，执行时不会被打断（仅适用于简单场合）

同步：一个线程访问数据未结束时，其他线程禁止对相同数据的访问。

锁：同步最常见的方法。

注解：锁是一种非强制机制，每一个线程在访问数据或资源之前首先试图获取（Acquire）锁，并在访问结束之后释放（Release）锁。当锁已经被占用的时候试图获取锁时，线程会等待，直到锁重新可用。

锁的分类：

1) 二元信号量（只有两种状态：占用和非占用，适合只能被唯一线程访问的资源）

2) 多元信号量，也叫信号量（一个初始值为N的信号量允许N个线程并发访问）

获取过程：

将信号量的值减1；

如果信号量的值小于0，则进入等待状态，否则继续执行。

释放过程：

将信号量的值加1；

如果信号量的值小于1，唤醒一个等待中的线程。

3) 互斥量（与二元信号量相似，资源仅允许被一个线程访问）

二元信号量与互斥量的区别：二元信号量：在整个系统可以被人以线程获取并释放，即，同一个信号量可以被系统中的一个线程获取后之后，可以由另一个线程释放；互斥量则要求哪个线程获取了互斥量，哪个线程就要负责释放这个锁。

4) 临界区（其作用范围仅限于本进程，其他进程无法获取该锁）

5) 读写锁（适用于读多，写少的场合）

读写锁有两种获取方式（共享的和独占的）

6) 条件变量（可以让许多线程一起等待某个事件的发生，当事件发生时，所有的线程可以一起恢复执行。多元信号量，只能让一个线程恢复执行。）

编译器的过度优化也可能造成线程安全的问题，看如下几个例子。

例1：

x = 0;

Thread1 Thread2

lock(); lock();

x++; x++;

unlock(); unlock();

由于有lock和unlock的保护，x++的行为不会被并发所破坏，那么x的值似乎必然是2了。其实不然，如果编译器为了提高x的访问速度，把x放到了某个寄存器里（不同线程的寄存器是各自独立的），因此如果Thread1先获得锁，则程序的执行可能是如下情况：

[Thread1]读取x的值到某个寄存器R[1]（R[1]=0）

[Thread1] R[1]++（由于之后可能还要访问x，因此Thread1暂时不讲R[1]写回x）

[Thread2]读取x的值到某个寄存器R[2]（R[2]=0）

[Thread2] R[2]++（R[2]=1）

[Thread2]将R[2]写回至x（x=1）

可见，在这样的情况下，即使正确的加锁，也不能保证多线程安全。

例2：

x = y = 0;

Thread1 Thread2

x = 1; y = 1;

r1 = y; r2 = x;

r1和r2至少有一个为1，逻辑上不可能同时为0.然而，事实上r1 = r2 = 0的情况确实可能发生。编译器在进行优化的时候，可能为了效率而交换毫不相干的两条相邻指令（如x=1和r1=y)的执行顺序。

则变为：

x = y = 0;

Thread1 Thread2

r1 = y; r2 = x;

x = 1; y = 1;

解决方法：

可以使用volatile来阻止过度优化，volatile主要做两件事

1）阻止编译器为了提高速度将一个变量缓存到寄存器内而不写回

2）阻止编译器调整操作volatile变量的指令顺序。

这个方法可以解决第一个问题，但不能完全解决第二个问题，因为volatile能够阻止编译器调整顺序，也无法阻止CPU动态调度换序。

例3：

volatile T* pInst = 0;

T* GetInstance()
{
if (pInst == NULL)
{
lock();
if(pInst == NULL)
pInst = new T;
unlock();
}
return pInst;
}

当函数返回时，pInst总是指向一个有效的对象。而lock和unclock防止了多线程竞争导致的麻烦。

然而，实际上这样的代码是有问题的，问题来源于CPU的乱序执行。

C++里的new操作包含两个步骤：

（1） 分配内存

（2） 调用构造函数

所以pInst = new T包含了三个步骤：

（1） 分配内存

（2） 在内存的位置上调用构造函数

（3） 将内存的地址赋给pInst

在这3步中，（2）（3）的顺序是可以颠倒的，也就是说可能出现这种情况：pInst的值已经不是NULL，但对象仍然没有构造完毕。这时候如果出现另外一个对GetInstance的并发调用，此时第一个 if内的表达式pInst == NULL为false，所以这个调用会直接返回尚未构造完全的对象的地址（pInst）以提供给用户使用。

从上面的例子中可以看出，阻止CPU换序是必需的。但目前并不存在可移植的阻止换序的方法。通常情况下是调用CPU提供的一条指令（barrier）。

barrier 指令用于阻止CPU将该指令之前的指令交换到barrier之后，反之亦然。

因此，例3可以修改为如下：

#define barrier() _asm_ volatile("lwsync")

volatile T* pInst = 0;

T* GetInstance()

{
if(!pInst)

{
lock();

if(!pInst)

{
T* temp = new T;

barrier();

pInst = temp;
}
unlock();
}
return pInst;
}

由于barrier的存在，对象的构造一定在barrier执行之前完成，因此，当pInst被赋值时，对象总是完好的。