C++11 并发编程教程 - Part 2 : 保护共享数据

上一篇文章咱们讲到如何启动一些线程去并发地执行某些操做，虽然那些在线程里执行的代码都是独立的，但一般状况下，你都会在这些线程之间使用到共享数据。一旦你这么作了，就面临着一个新的问题 —— 同步。

   下面让咱们用示例来阐释“同步”是个什么问题。

同步问题

   咱们就拿一个简单的计数器做为示例吧。这个计数器是一个结构体，他拥有一个计数变量，以及增长或减小计数的函数，看起来像这个样子：

   [译注：原文 Counter 的 value 并未初始化，其初始值随机，读者可自行初始化为 0 ]

1 2 3 4 5 6

struct  Counter {     int  value;     void  increment(){         ++value;     } };

   这并没什么稀奇的，下面让咱们来启动一些线程来增长计数器的计数吧。

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

int  main(){     Counter counter;     std::vector<std::thread> threads;     for(int  i = 0; i < 5; ++i){         threads.push_back(std::thread([&counter](){             for(int  i = 0; i < 100; ++i){                 counter.increment();             }         }));     }     for(auto&  thread  : threads){         thread.join();     }     std::cout << counter.value << std::endl;     return  0; }

   [译注：bill的测试环境下，上述代码始终输出 500，读者可将外层 for 循环条件改成 i < 100，内层for 循环条件改成 i < 99999 以观察实验结果]

   一样的，也没什么新花样，咱们只是启动了 5 个线程，每一个线程都让计数器增长 100 次而已。等这一工做结束，咱们就打印计数器最后的数值。

   若是运行这一程序，咱们理所固然的指望运行结果是 500，但事与愿违，没人能保证这个程序最终输出什么。下面是在个人机器上获得的一些结果：

1 2 3 4 5 6

442 500 477 400 422 487

   问题的根源在于计数器的 increment() 并不是原子操做，而是由 3 个独立的操做组成的：

       1. 读取 value 变量的当前值。

       2. 将读取的当前值加 1。

       3. 将加 1 后的值写回 value 变量。

   当你以单线程运行上述代码时，就不会出现任何问题，上述三个步骤会按照顺序依次执行。可是一旦你身处多线程环境，状况就会变得糟糕起来，考虑以下执行顺序：

       1. 线程a：读取 value 的当前值，获得值为 0。加1。所以 value = 1。[译注：此时 1 并无写回value 内存，原文“value = 1”仅做逻辑意义，下同]

       2. 线程b：读取 value 的当前值，获得值为 0。加1。所以 value = 1。

       3. 线程a：将 1 写回 value 内存并返回 1。

       4. 线程b：将 1 写回 value 内存并返回 1。

   这种状况源于线程间的 interleaving。Interleaving 描述了多线程同时执行几句代码的各类状况。就算仅仅只有两个线程同时执行这三个操做，也会存在不少可能的 interleaving。当你有许多线程同时执行多个操做时，要想枚举出全部 interleaving，几乎是不可能的。并且若是线程在执行单个操做的不一样指令之间被抢占，也会致使 interleaving 的发生。

   目前有许多能够解决这一问题的方案：

• Semaphores

• Atomic references

• Monitors

• Condition codes

• Compare and swap

• etc.

   就本文而言，咱们将学习如何使用 Semaphores 去解决这一问题。事实上，咱们仅仅使用了Semaphores 中比较特殊的一种 —— 互斥量。互斥量是一个特殊的对象，在同一时刻只有一个线程可以获得该对象上的锁。借助互斥量这种简而有力的性质，咱们即可以解决线程同步问题。

使用互斥量保证 Counter 的线程安全

   在 C++11 的线程库中，互斥量被放置于头文件 <mutex>，并以 std::mutex 类加以实现。互斥量有两个重要的函数：lock() 和 unlock()。顾名思义，前者使当前线程尝试获取互斥量的锁，后者则释放已经获取的锁。lock() 函数是阻塞式的，线程一旦调用 lock()，就会一直阻塞直到该线程得到对应的锁。

   为了使咱们的计数器具有线程安全性，咱们须要对其添加 std::mutex 成员，并在成员函数中对互斥量进行 lock()/unlock() 调用。

1 2 3 4 5 6 7 8 9 10

struct  Counter {     std::mutex mutex;     int  value;     Counter() : value(0) {}     void  increment(){         mutex.lock();         ++value;         mutex.unlock();     } };

   若是咱们如今再次运行以前的测试程序，咱们将始终获得正确的输出：500。

异常与锁

如今让咱们来看看另一种状况会发生什么。假设如今咱们的计数器拥有一个 derement() 操做，当  value 被减为 0 时抛出一个异常： 

1 2 3 4 5 6 7 8 9 10 11 12 13 14

struct  Counter {     int  value;                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                 Counter() : value(0) {}     void  increment(){         ++value;     }     void  decrement(){         if(value == 0){             throw  "Value cannot be less than 0";         }         --value;     } };

   假设你想在不更改上述代码的前提下为其提供线程安全性，那么你须要为其建立一个 Wrapper 类：

1 2 3 4 5 6 7 8 9 10 11 12 13 14

struct  ConcurrentCounter {     std::mutex mutex;     Counter counter;     void  increment(){         mutex.lock();         counter.increment();         mutex.unlock();     }     void  decrement(){         mutex.lock();         counter.decrement();           mutex.unlock();     } };

   这个 Wrapper 将在大多数状况下正常工做，然而一旦 decrement() 抛出异常，你就遇到大麻烦了，当异常被抛出时，unlock() 函数将不会被调用，这将致使本线程得到的锁不被释放，你的程序也就瓜熟蒂落的被永久阻塞了。为了修复这一问题，你须要使用 try/catch 块以保证在抛出任何异常以前释放得到的锁。

1 2 3 4 5 6 7 8 9 10

void  decrement(){     mutex.lock();     try  {         counter.decrement();     }  catch  (std::string e){         mutex.unlock();         throw  e;     }     mutex.unlock(); }

   代码并不复杂，可是看起来却很丑陋。试想一下，你如今的函数拥有 10 个返回点，那么你就须要在每一个返回点前调用 unlock() 函数，而忘掉其中的某一个的可能性是很是大的。更大的风险在于你又添加了新的函数返回点，却没有对应地添加 unlock()。下一节将给出解决此问题的好办法。

锁的自动管理

   当你想保护整个代码段（就本文而言是一个函数，但也能够是某个循环体或其余控制结构[译注：即一个做用域]）免受多线程的侵害时，有一个办法将有助于防止忘记释放锁：std::lock_guard。

   这个类是一个简单、智能的锁管理器。当 std::lock_guard 实例被建立时，它自动地调用互斥量的lock() 函数，当该实例被销毁时，它也顺带释放掉得到的锁。你能够像这样使用它：

1 2 3 4 5 6 7 8 9 10 11 12

struct  ConcurrentSafeCounter {     std::mutex mutex;     Counter counter;     void  increment(){         std::lock_guard<std::mutex> guard(mutex);         counter.increment();     }     void  decrement(){         std::lock_guard<std::mutex> guard(mutex);         counter.decrement();     } };

   代码变得更整洁了不是吗？

   使用这种方法，你无须绷紧神经关注每个函数返回点是否释放了锁，由于这个操做已经被std::lock_guard 实例的析构函数接管了。

总结

   如今咱们结束了短暂的 Semaphores 之旅。在本章中你学习了如何使用 C++ 线程库中的互斥量来保护你的共享数据。

   但有一点请牢记：锁机制会带来效率的下降。的确，一旦使用锁，你的部分代码就变得有序[译注：非并发]了。若是你想要设计一个高度并发的应用程序，你将会用到其余一些比锁更好的机制，但他们已不属于本文的讨论范畴。

下篇

   在本系列的下一篇文章中，我将谈及关于互斥量的一些进阶概念，并介绍如何使用条件变量去解决一些并发编程问题。