一步一步写线程之十二无锁编程

一、无锁编程

无锁编程并不是真正的无锁，只是在软件上消除了锁（或者说消除了传统认知中的锁）。牺牲CPU的占用时间来换取效率。无论是传统的单线程编程还是后来的多线程编程及至并发编程，其实抽象出来的模型就是生产者和消费者。这种生产者和消费者的模型，在很多情况下其实是不需要锁也不需要缓冲队列之类的，各自干各自的活就可以。但事情总不都是按照想象进行的，总有一些生产者和消费者的任务是不匹配的，在这种情况下，如何更好的从算法层次上将生产者和消费者的匹配度做的更好，就是一个重要的问题。
体现到现实世界来，就是提高工作效率，降低工作成本。生产者和消费者有1：1,1：N，还有N：1和N：N几种关系。在生产者和消费者能力不对等的情况下，就需要一个缓冲区（一般是以队列来实现）来存储工作任务，这时，任务的生产和消费端都需要进行锁的控制，这是前面学习的一个重要的知识点。
但是，有没有一种办法，可以让任务向队列缓冲区插入和读取时，不使用锁呢？答案就是使用无锁编程。

二、无锁编程的要点

1、原子编程
原子编程其实就是原子操作的行为，这对于大家可能已经很熟悉了，会用不会用放一边，肯定耳朵都听得长茧子了。原子操作非常容易理解，就是原子不可分割，那么这个操作也是必须一气呵成，不能被多线程或者其它类似的任务打断。那么这就保证了数据的完整性和一致性。在c++特别是c++11后的新标准中，提供了大量的原子操作的类型如std::atomic。
2、内存序
内存序其实是一种指令操作的约定或者说标准，用来匹配硬件与上层软件为达到处理顺序维持一致性的前提。而在多线程和分布式编程中，经常会遇到这种情况，特别是不同的架构CPU及不同的操作系统中，这种更是经常遇到的。
其实无锁队列本身就是一种特殊的生产者和消费者，而无锁队列的实现，对整体行为操作的原子性和顺序性提出了严格的要求。一般来说，实现这种机制的最基础的方法是使用内存屏障。内存屏障又可以分为内存屏障和编译器屏障，听名字就可以知道，它们主要是用来防止编译器或处理器进行指令重排，保持多线程环境下的数据一致性。内存屏障其实是一种计算机上的抽象的概念，具体的不同的平台和语言都有自己的实现机制，在c++中就各种的锁和原子操作等。

3、CAS的ABA问题
CAS，即Compare-And-Swap，比较和交换，其实就是大家认知里的无锁编程的基础技术。它采用了一种原子操作加CPU循环等待的方式来实现安全的数据读写。在c++中，提供了几个重要的CAS的接口函数，如compare_exchange_strong和compare_exchange_weak等。
可能许多开发者听到无锁编程会眼前一亮，心想总算扔掉了锁这个包袱，且先不要乐观。基于CAS的无锁编程，确实有不少优点，但这里面缺点也不少。这个在前面分析过。这里面有一个很让人难受的问题，那就是ABA问题。什么是ABA问题呢？举个例子就明白了，有一个共享变量的值是10，线程1修改其为11，然后线程2修改其为12，随后又修改其为11，此时线程1再操作此变量时，发现其没有改变。则其随后针对的处理行为可能就会出现问题。对，换句话说，ABA问题，只有在场景中需要处理时才有意义，否则可以忽略。而这种需要处理的场景，往往是涉及到资产的情况，那么这就非常重要了。
如何解决ABA问题呢？一般来说，是使用版本号或者打时间戳等方式来解决。ABA问题的本质就是放弃使用锁导致的线程自由处理共享变量付出的代价。

4、优化处理
没有任何一种手段是普适的。无锁队列也是如此，所以没有最好，只有最合适即针对实际的场景进行优化。无锁队列的优化非常复杂，它不但涉及到传统的优化问题，如一些指令并发、OS的API控制等等，还要处理对循环等待的时间，次数是否处理失败的情况以及减少误操作的机会等等。特别是在分布式编程中，更是复杂，这就需要系统的掌握相关的底层知识和开发技术等。

5、无锁编程的API
说到无锁编程，其实这个在各个平台都有自己提供的接口。在c++编程中提供了：

bool compare_exchange_weak( T& expected, T desired,std::memory_order success,std::memory_order failure ) noexcept;
bool compare_exchange_weak( T& expected, T desired,std::memory_order success,std::memory_order failure ) volatile noexcept;
bool compare_exchange_weak( T& expected, T desired,std::memory_order order =std::memory_order_seq_cst ) noexcept;
bool compare_exchange_weak( T& expected, T desired,std::memory_order order =std::memory_order_seq_cst ) volatile noexcept;
bool compare_exchange_strong( T& expected, T desired,std::memory_order success,std::memory_order failure ) noexcept;
bool compare_exchange_strong( T& expected, T desired,std::memory_order success,std::memory_order failure ) volatile noexcept;
bool compare_exchange_strong( T& expected, T desired,std::memory_order order =std::memory_order_seq_cst ) noexcept;
bool compare_exchange_strong( T& expected, T desired,std::memory_order order =std::memory_order_seq_cst ) volatile noexcept;

强弱二者的不同在于针对不同的架构处理器，weak允许出现偶然的错误返回。这样做的目的只有一个，在某些情况下可能效率会更高。
而LINUX GNUC标准中提供了：

bool __atomic_compare_exchange_n(type *ptr,              // 比较的ptrtype *expected,         // 旧值，返回ptr指向的值type desired,           // 设置的新值bool weak,              // 强一致或弱一致int success_memorder,   // 成功时内存序int failure_memorder    // 失败时内存序
)

在Windows中标准提供了：

//32位
LONG InterlockedCompareExchange([in, out] LONG volatile *Destination,//指向目标值的指针[in]      LONG          ExChange,//交换值。[in]      LONG          Comperand//要与 Destination 进行比较的值
);//函数将 Destination 值与 Compareand 值进行比较。 如果 Destination 值等于 Compareand 值， 则 Exchange 值将存储在 Destination 指定的地址中。 否则，不会执行任何操作。
//64位
LONG64 InterlockedCompareExchange64([in, out] LONG64 volatile *Destination,[in]      LONG64          ExChange,[in]      LONG64          Comperand
);
//指针
PVOID InterlockedCompareExchangePointer([in, out] PVOID volatile *Destination,[in]      PVOID          Exchange,[in]      PVOID          Comperand
);

Windows的相关设置比较简单而且其文档也比较全，这也是微软提供的文档相对丰富原因。

三、无锁编程的应用场景

无锁编程听起来比有锁编程要好很多啊，肯定要包打天下啊。可事实并不是，换句话说，无锁编程也是有其的应用场景的。首先需要了解一下有锁编程缺点：
1、线程切换引起的Cache失效
2、阻塞引起的线程切换和线程休眠
3、内存分配时的锁导致的性能下降
而无锁编程就是有针对性的通过CPU忙等待（牺牲CPU时间）而不是休眠线程来换取1和2的缓解的。所以无锁编程适应场景也就出来了：
1、读写操作频繁，一般建议在十万量级以上（忙等时间尽量小），至少也得在万级才可以考虑
2、读写操作耗时尽量短或者说任务无阻塞操作（忙等时间尽量小）
而对于3,无锁编程其实和有锁编程解决的方式是类似的。或者直接分配好的数组或者使用内存池。
从上面可以再次印证一个事实，只有最合适，没有最优。一切都是平衡的结果。

四、无锁编程的实现机制和实际库应用

无锁编程的实现其实主要有两种，一种是基于链表的实现，这种在资料中非常容易找到。通常是一个链表来模拟实现无锁的读写；另外一个就是使用数组。当然，既然二者都可以实现，那么混合着也可以实现。这个看开发者个人的喜好的实际的情况。
而在实际应用中，包括许多技术大牛，研究论文和有名的框架都对无锁编程进行了阐述和分析，并给出了相关的实现代码。比如：《Implementing Lock-Free Queues》和《Simple, Fast, and Practical Non-Blocking and Blocking ConcurrentQueue Algorithms》等。而框架实现中常见的有intel tbb，folly和boost。比如boost中的lockfree::queue 和lockfree::stack等。而folloy中则提供了AtomicIntrusiveLinkedList等。至于tbb，无锁搞得还是相当好，有兴趣自己下载源码分析即可。
说明：网上有太多的无锁编程的例子，大家要仔细分析，去芜存菁，不要乱了阵脚。

五、简单的实现

先看一下简单的CAS的编程：

#include <atomic>template<typename T>
struct node
{T data;node* next;node(const T& data) : data(data), next(nullptr) {}
};template<typename T>
class stack
{std::atomic<node<T>*> head;
public:void push(const T& data){node<T>* new_node = new node<T>(data);// put the current value of head into new_node->nextnew_node->next = head.load(std::memory_order_relaxed);// now make new_node the new head, but if the head// is no longer what's stored in new_node->next// (some other thread must have inserted a node just now)// then put that new head into new_node->next and try againwhile (!head.compare_exchange_weak(new_node->next, new_node,std::memory_order_release,std::memory_order_relaxed)); // the body of the loop is empty// Note: the above use is not thread-safe in at least
// GCC prior to 4.8.3 (bug 60272), clang prior to 2014-05-05 (bug 18899)
// MSVC prior to 2014-03-17 (bug 819819). The following is a workaround:
//      node<T>* old_head = head.load(std::memory_order_relaxed);
//      do
//      {
//          new_node->next = old_head;
//      }
//      while (!head.compare_exchange_weak(old_head, new_node,
//                                         std::memory_order_release,
//                                         std::memory_order_relaxed));}
};int main()
{stack<int> s;s.push(1);s.push(2);s.push(3);
}

再看一个 strong CAS：

#include <atomic>
#include <iostream>std::atomic<int> ai;int tst_val = 4;
int new_val = 5;
bool exchanged = false;void valsout()
{std::cout << "ai = " << ai<< "  tst_val = " << tst_val<< "  new_val = " << new_val<< "  exchanged = " << std::boolalpha << exchanged<< '\n';
}int main()
{ai = 3;valsout();// tst_val != ai   ==>  tst_val is modifiedexchanged = ai.compare_exchange_strong(tst_val, new_val);valsout();// tst_val == ai   ==>  ai is modifiedexchanged = ai.compare_exchange_strong(tst_val, new_val);valsout();
}

无锁编程的实现其实并没有想象中的难，但之所以大家感觉到有些难的原因不外乎两个：一个是应用的场景对大多数程序员来说不存在，也就是用得少；第二个就是无锁编程与数据结构、算法和OS甚至编译等基础知识都有较强的相关性，即使想用好可能也需要补充很多相关的技术。
这里只是把无锁编程进行一个简单的例程说明，无锁队列的实现，在后期再完善。

六、总结

这里需要声明一个问题，就是在生产者消费者绝对不对等的情况下，使用何种算法和技巧都是没有办法解决问题的。这句话是什么意思呢？就是说十个人干得活，如果只有一个小孩来干，用什么方法在当前状态下也是无解的。但如果合理的采用一些调度算法，安排一下流程，可能五个人就可以完成十个人的工作量。这是不是有点象武学上的“一力降十会”？
同样，无锁队列的目的是提高效率，而不是从解决不可能解决的问题。大家千万不要走进误区，切记！