无锁编程技术及实现「建议收藏」

大家好，又见面了，我是你们的朋友全栈君。

1.基于锁的编程的缺点

多线程编程是多CPU系统在中应用最广泛的一种编程方式，在传统的多线程编程中，多线程之间一般用各种锁的机制来保证正确的对共享资源（share resources）进行访问和操作。

在多线程编程中只要需要共享某些数据，就应当将对它的访问串行化。比如像++count(count是整型变量)这样的简单操作也得加锁，因为即便是增量操作这样的操作，，实际上也是分三步进行的：读、改、写（回）。

movl x, %eax

addl $1, %eax
movl %eax, x

更进一步，甚至内存变量的赋值操作都不能保证是原子的，比如在32位环境下运行这样的函数

void setValue()

{

value = 0x100000006;

}

执行的过程中，这两条指令之间也是可以被打断的，而不是一条原子操作。（也就是所谓的写撕裂）

所以修改共享数据的操作必须以原子操作的形式出现，这样才能保证没有其它线程能在中途插一脚来破坏相应数据。

而在使用锁机制的过程中，即便在锁的粒度（granularity），负载（overhead），竞争（contention），死锁（deadlock）等需要重点控制的方面解决的很好，也无法彻底避免这种机制的如下一些缺点：

1，锁机制会引起线程的阻塞（block），对于没有能占用到锁的线程或者进程，将一直等待到锁的占有者释放锁资源后才能继续执行，而等待时间理论上是不可设置和预估的。

2，申请和释放锁的操作，增加了很多访问共享资源的消耗，尤其是在锁竞争(lock-contention)很严重的时候，比如这篇文章所说：http://preshing.com/20111118/locks-arent-slow-lock-contention-is/

3，现有实现的各种锁机制，都不能很好的避免编程开发者设计实现的程序出现死锁或者活锁的可能

4，优先级反转（prorithy inversion）和锁护送(Convoying)的现象

5，难以调试

无锁编程（Lock-Free）就是在某些应用场景和领域下解决以上基于锁机制的并发编程的一种方案。

2.无锁编程(LOCK-FREE)的定义

提到无锁编程（lock-free），按字面最直观的理解是不使用锁的情况下实现多线程之间对变量同步和访问的一种程序设计实现方案。严格的说这个理解是不对的，Lock-Free的程序肯定是不包括锁机制的，而不包括锁机制的程序不一定是lock-free的。

更准确的说，在并发编程上按照同步的维护划分，可以分为阻塞的编程方式（Block）和非阻塞的编程方式（Non-blocking Synchronization）。阻塞的编程方式基本是基于锁的（lock-based）。其中无锁编程（Lock-free）属于非阻塞同步（Non-blocking Synchronization）中的一种情况，实现非阻塞同步的算法方案按照效果要求不同可以粗略的分为：

Wait-free: 满足等待无关的程序，任何线程可以在有限步之内结束，不管其它线程的执行速度和进度如何

Lock-free：锁无关的程序，一个锁无关的程序能够确保它所有线程中至少有一个能够继续往下执行，而有些线程可能会被的延迟。然而在整体上，在某个时刻至少有一个线程能够执行下去。作为整体进程总是在前进的，尽管有些线程的进度可能没有其它线程进行的快。

Obstruction-free：在任何时间点，一个孤立运行线程的每一个操作可以在有限步之内结束。只要没有竞争，线程就可以持续运行。一旦共享数据被修改，Obstruction-free 要求中止已经完成的部分操作进行回滚。

更细致的还可以把并发编程按效果划分为:

Blocking

1. Blocking

2. Starvation-Free

Obstruction-Free

3. Obstruction-Free

Lock-Free

4. Lock-Free (LF)

Wait-Free

5. Wait-Free (WF)

6. Wait-Free Bounded (WFB)

7. Wait-Free Population Oblivious (WFPO)

具体细节可以参考这篇文章http://ifeve.com/lock-free-and-wait-free/，有对阻塞非阻塞的等定义的详细描述，这里不详细论述。

3.无锁编程中涉及的一些技术原理

无锁编程具体使用和考虑到的技术方法包括：原子操作（atomic operations）, 内存栅栏（memory barriers）, 内存顺序冲突（memory order），指令序列一致性（sequential consistency）和顺ABA现象等等，这方面借用一篇资料的总结的图概况：

无锁编程技术及实现「建议收藏」

在这其中最基础最重要的是操作的原子性或说原子操作。原子操作可以理解为在执行完毕之前不会被任何其它任务或事件中断的一系列操作。原子操作是非阻塞编程最核心基本的部分，没有原子操作的话，操作会因为中断异常等各种原因引起数据状态的不一致从而影响到程序的正确。

对于原子操作的实现机制，在硬件层面上CPU处理器会默认保证基本的内存操作的原子性，CPU保证从系统内存当中读取或者写入一个字节的行为肯定是原子的，当一个处理器读取一个字节时，其他CPU处理器不能访问这个字节的内存地址。但是对于复杂的内存操作CPU处理器不能自动保证其原子性，比如跨总线宽度或者跨多个缓存行（Cache Line），跨页表的访问等。这个时候就需要用到CPU指令集中设计的原子操作指令，现在大部分CPU指令集都会支持一系列的原子操作。而在无锁编程中经常用到的原子操作是Read-Modify-Write （RMW）这种类型的，这其中最常用的原子操作又是 COMPARE AND SWAP（CAS），几乎所有的CPU指令集都支持CAS的原子操作，比如X86平台下中的是 CMPXCHG。

继续说一下CAS，CAS操作行为是比较某个内存地址处的内容是否和期望值一致，如果一致则将该地址处的数值替换为一个新值。CAS能够操作的位数越多，使用它来实现锁无关的数据结构就越容易（细节可以在intel手册中查看）。CAS操作具体的实现原理主要是两种方式：总线锁定和缓存锁定。所谓总线锁定，就是CPU执行某条指令的时候先锁住数据总线的，使用同一条数据总线的CPU就无法访问内存了，在指令执行完成后再释放锁住的数据总线。锁住数据总线的方式系统开销很大，限制了访问内存的效率，所以又有了基于CPU缓存一致性来保持操作原子性作的方法作为补充，简单来说就是用CPU的缓存一致性的机制来防止内存区域的数据被两个以上的处理器修改（可详见CPU缓存的MESI协议）。

在操作系统的层面，Linux系统提供了软件级的原子操作，包括两大类系统调用，一类是基于对整数进行操作的atomic_set/and/inc，一类是针对单独的位进行操作的set/clear/change_bit，它们大部分都是基于硬件层面的CAS的指令实现的。

在各种开发语言中（c,c++,java）基于操作系统提供的接口也都封装实现了对应的原子操作api，所以开发者完全可以直接调用各个开发语言提供的接口实现无锁程序。

除了使用原子操作保证操作的原子性，还要考虑在不用的语言和内存模型下，如何保证，操作的顺序性，编译时和运行时的指令重排序，还有由假共享引起内存顺序冲突（Memory order violation）等等细节。原子性确保指令执行期间不被打断，要么全部执行，要么根本不执行，而顺序性确保即使两条或多条指令出现在独立的执行线程中或者独立的处理器上时，保持它们本该执行的顺序。假共享是指多个CPU同时修改同一个缓存行的不同部分而引起其中一个CPU的操作无效，当出现这个内存顺序冲突时，CPU必须清空流水线。这些其实在多核编程的环境下的locked-base的实现中也有类似的问题所以这里就不多展开。

4.几个例子

John D. Valois 《Implementing Lock-Free Queues》中提到的一个基于链表的无锁队列链表的实现，可以作为了解lock-free一个例子

EnQueue(x) { //入队列方法

q = new record();

q->value = x; //队列节点的值

q->next = NULL;//下一个节点

p = tail; //保存尾节点指针

oldp = p;

do { //开始 loop cas

while (p->next != NULL) //用来防止进行cas（tail,oldp,q）操作的线程挂掉引起死循环

p = p->next;

} while( CAS(p.next, NULL, q) != TRUE);

CAS(tail, oldp, q);

}

DeQueue() //出队列方法

{

do{

p = head;

if (p->next == NULL){

return ERR_EMPTY_QUEUE;

}

while( CAS(head, p, p->next) != TRUE );

return p->next->value;

}

具体实现中，在linux 平台上gcc编译器（内置了支持cas操作的函数，或者直接调用汇编命令（比如CMPXCHG 和 CMPXCHG8B）也可以在c程序里实现cas操作。GCC (GNU Compiler Collection，4.1 和更高版本）提供几个内置函数，可以使用它们在 x86 和 x86-64 平台上实现 CAS 操作，这些内置函数包括：

type __sync_fetch_and_add (type *ptr, type value)

type __sync_fetch_and_sub (type *ptr, type value)

type __sync_fetch_and_or (type *ptr, type value)

type __sync_fetch_and_and (type *ptr, type value)

type __sync_fetch_and_xor (type *ptr, type value)

type __sync_fetch_and_nand (type *ptr, type value)

type __sync_add_and_fetch (type *ptr, type value)

type __sync_sub_and_fetch (type *ptr, type value)

type __sync_or_and_fetch (type *ptr, type value)

type __sync_and_and_fetch (type *ptr, type value)

type __sync_xor_and_fetch (type *ptr, type value)

type __sync_nand_and_fetch (type *ptr, type value)

bool __sync_bool_compare_and_swap (type *ptr, type oldval type newval, …)

type __sync_val_compare_and_swap (type *ptr, type oldval type newval, …)

type __sync_lock_test_and_set (type *ptr, type value, …)

更多细节分析可参看John D. Valois的《Implementing Lock-Free Queues》

同样，在java语言中Lock-Free的数据结构和算法其实更加常见，在java.util.concurrent包中大量实现都是建立在基于CAS实现Lock-Free算法上，没有CAS就不会有此包。Java.util.concurrent.atomic提供了基于CAS实现的若干原语：

AtomicBoolean — 原子布尔

AtomicInteger — 原子整型

AtomicIntegerArray — 原子整型数组

AtomicLong — 原子长整型

AtomicLongArray — 原子长整型数组

AtomicReference — 原子引用

AtomicReferenceArray — 原子引用数组

AtomicMarkableReference — 原子标记引用

AtomicStampedReference — 原子戳记引用

AtomicIntegerFieldUpdater — 用来包裹对整形 volatile 域的原子操作

AtomicLongFieldUpdater — 用来包裹对长整型 volatile 域的原子操作

AtomicReferenceFieldUpdater — 用来包裹对对象 volatile 域的原子操作

引入这些类的主要目的就是为了实现Lock-Free算法和相关数据结构，以incrementAndGet这个方法为例：

public final int incrementAndGet() {

for (;;) {

int current = get();

int next = current + 1;

if (compareAndSet(current, next))

return next;

}

在这里使用CAS原子操作，每次读取数据数值后将此数值和+1后的结果进行CAS操作，如果成功就返回结果，否则重试到成功为止。其中compareAndSet是java中实现的CAS函数，在java语言中的实现，是借助JNI机制来调用汇编实现的：

public final boolean compareAndSet(int expect, int update) {

return unsafe.compareAndSwapInt(this, valueOffset, expect, update);

}

unsafe.compareAndSwapInt是个本地方法调用，对应的x86处理器的jni源码

#define LOCK_IF_MP(mp) __asm cmp mp, 0 \

__asm je L0 \

__asm _emit 0xF0 \

__asm L0:

inline jint Atomic::cmpxchg (jint exchange_value, volatile jint* dest, jint compare_value) {

// alternative for InterlockedCompareExchange

int mp = os::is_MP();

__asm {

mov edx, dest

mov ecx, exchange_value

mov eax, compare_value

LOCK_IF_MP(mp)//判断是否是多核，是则添加LOCK指令维护顺序一致性

cmpxchg dword ptr [edx], ecx

}

再看看在java里无锁队列的实现作为对比参照：

import java.util.concurrent.atomic.AtomicReference;

public class LockFreeQueue<T> {

private AtomicReference<Node> head; //可以规避ABA现象

private AtomicReference<Node> tail;

/**

* Create a new object of this class.

public LockFreeQueue() {

Node sentinel = new Node(null);

　 head = new AtomicReference<Node>(sentinel);

　 tail = new AtomicReference<Node>(sentinel);

}

/**

* Enqueue an item.

* @param value Item to enqueue.

public void enq(T value) {

　 // try to allocate new node from local pool

　 Node node = new Node(value);

　 while (true) {
　　　　　　　

　　 Node last = tail.get();　

　　 Node next = last.next.get();

　　 // are they consistent

　　 if (last == tail.get()) {

　　　 if (next == null) {
　　

　　　　 // try to link node to end of list

　　　　 if (laspareAndSet(next, node) {

　　　　　 // enq done, try to advance tail

　　　　　 tail.compareAndSet(last, node);

　　　　　 return;

　　　　 }

　　　 } else {
　　　　　　　

　　　　 // try to swing tail to next node

　　　　 tail.compareAndSet(last, next);

　　　 }

/**

* Dequeue an item.

* @throws queue.EmptyException The queue is empty.

* @return Item at the head of the queue.

public T deq() throws EmptyException {

　 while (true) {

　　 Node first = head.get();

　　 Node last = tail.get();

　　 Node next = first.next.get();

　　 // are they consistent

　　 if (first == head.get()) {

　　　 if (first == last) {
　

　　　　 if (next == null) {
　

　　　　　 throw new EmptyException();

　　　　 }

　　　　 // tail is behind, try to advance

　　　　 tail.compareAndSet(last, next);

　　　 } else {

　　　　 T value = next.value;

　　　　 if (head.compareAndSet(first, next)) {

　　　　　 return value;

　　　　 }

/**

* Items are kept in a list of nodes.

public class Node {

　 /**

　　 * Item kept by this node.

　 public T value;

　 /**

　　 * Next node in the queue.

　 public AtomicReference<Node> next;

　　/**

　　 * Create a new node.

　 public Node(T value) {

　　 this.next = new AtomicReference<Node>(null);

　　 this.value = value;

　 }

最后这里随便说一下CAS操作的ABA的问题，所谓的ABA的问题简要的说就是，线程a先读取了要对比的值v后，被线程b抢占了，线程b对v进行了修改后又改会v原来的值，线程1继续运行执行CAS操作的时候，无法判断出v的值被改过又改回来。ABA对基于指针实现的算法影响很大。上面的C语言里实现的程序是有可能被ABA问题影响的，因为它的CAS操作判断的是指针的地址，这个地址被重用的可能性很大（地址被重用是很经常发生的，一个内存分配后释放了，再分配，很有可能还是原来的地址，内存管理中重用内存基本上是一种很常见的行为）。解决ABA的问题的一种方法是，一次用CAS检查双倍长度的值，前半部是指针，后半部分是一个计数器；或者对CAS的数值加上版本号。《Implementing Lock-Free Queues》也有提到相关的解决方案。

5.结论和建议

无锁编程方式相对基于锁的编程方式，具备一定的优点，比如不会发生死锁，不会有优先级倒置，进行CAS操作的消耗比加锁操作轻很多等等。单从这个角度上讲在对应用程序不太复杂，而对操作实时性要求较高时，采用无锁多线程能发挥一定优势。

在性能上基于CAS实现的硬件级的互斥，其单次操作性能比相同条件下的应用层的较为高效，但当多个线程并发时，硬件级的互斥引入的消耗一样很高(类似spin_lock)。无锁算法及相关数据结构并不意味在所有的环境下都能带来整体性能的极大提升。循环CAS操作对时会大量占用cpu，对系统时间的开销也是很大。这也是基于循环CAS实现的各种自旋锁不适合做操作和等待时间太长的并发操作的原因。而通过对有锁程序进行合理的设计和优化，在很多的场景下更容易使程序实现高度的并发性。

在开发维护的成本和复杂度上，无锁编程难度非常高，类似ABA的问题也比较难直观的探测和解决，并且实现细节和硬件平台相关性很强。目前理论和实践上只有少数数据结构可以支持实现无锁编程，比如队列、栈、链表、词典等，目前要在产品业务开发环境中进行大规模的无锁编程较为困难，更多的是用在部分特殊场景解决锁竞争等问题比较合适，比如操作系统中实现metux，semaphare, 一些语言的库的实现（比如 java current lib， lmax disprute）。若想在应用开发中尝试的Lock-Free的方案，建议可以选择合适的第三方lib实现。

附一. 些常见的相关lib库工具

https://github.com/mthssdrbrg/LockFreeStack

http://www.lmax.com/disruptor

http://www.cse.chalmers.se/research/group/noble/

http://www.cl.cam.ac.uk/research/srg/netos/lock-free/

http://www.rossbencina.com/code/lockfree

http://www.liblfds.org/

http://concurrencykit.org/

http://www.cl.cam.ac.uk/research/srg/netos/lock-free/

附二. 参考资料和进阶阅读;

Writing Lock-Free Code: A Corrected Queue

The Trouble With Locks