C++笔记之多线程的理解与应用

一、线程与进程

进程，拥有资源并且独立运行的基本单位；

将CPU比作是是一个工厂，那么进程可以看做是一个可以独立进行工作的车间，车间内有生产必须的资源以及工人，同时工厂内同一时刻只有一个车间在开工，但是工厂内是可以有多个车间的。[1]

线程，程序执行的最小单元；

线程则是车间内的工人，工人的数量可以有多个，某些工具或者空间只有一个，需要多人分享（共享资源），如果有一个人正在使用，那么其他工人则必须等待此人使用完毕。

进程拥有独立的执行环境，每个进程都有自己的内存空间。进程间的通信可以使用管道或者socket等。在一个进程中，线程共享该进程的资源，比如内存或者打开的文件。由于进程之间是隔离的，而线程之间共享资源，所以线程之间存在竞争。

二、线程的状态

1、就绪

线程具备运行的所有条件，等待处理；

2、运行

线程占了CPU资源，正在执行

3、阻塞

线程等待一个事件或者信号量，此时线程无法执行

三、并发与并行

并发是同一时间应对多件事情的能力；而并行则是同一时间做多件事情的能力。

并行就是在相同的时间，多个工作执行单位同时执行；

在单核CPU上，多线程本质上是单个 CPU 的时间分片，一个时间片运行一个线程的代码，它可以支持并发处理，但是不能说是真正的并行计算。

而在多核的CPU上，则实现了真正意义上的并行。

多线程如下图所示：



四、多线程安全

多线程并发会存在以下三个方面的问题：

（1）最首要的是安全问题：

安全主要在两个方面：

一是两个线程之间的相互干扰；

当多个线程对同一个数据进行操作的时候，如果多个线程之间出现了交错。

比如a++：

首先获取a的值，然后对其加1，最后将更新的值保存在a中；

同样的操作对于b–；

此时有两个线程对a分别进行上述操作，有可能出现的是：

线程一获取a值，线程二获取a值；

线程一对a进行+1，线程二对a进行-1；

线程一更新a值，线程二更新a值；

最后a值没有改变。

二是内存一致性，也就是不同的线程对同一数据产生了不同的看法。

假设

int a = 0；

线程一进行自增操作：

a++；

然后，线程二进行输出操作：

c<<a ;

最后输出的结果既有可能是0也有可能是1了。

解决方案：

为了解决上述问题，最直接的方法是互斥锁，就是当一个线程对某个数据进行排他性访问的时候，会得到了其内部锁之后才能访问，而在该线程没有释放此锁的情况下，其他的线程是无法访问的。

另外一种方式则是阻塞block，当一个线程在运行时，另一个线程处于休眠的状态。总而言之，就是将并行转换为串行来避开竞争的问题。

你以为加了锁就可以高枕无忧了吗，naive

因为互斥锁或者阻塞又会引入新的问题——

（2）线程活跃度

首先， 当一个线程锁定了另一个线程需要的资源的时候，而第二个线程又锁定了第一个线程需要的资源，这个时候就会发生死锁，双方都在等待对方先释放所需资源。

其次，当一个线程长时间地占用某个资源，而其他的线程只能被迫长时间地等待。

然后，线程1可以使用资源，但是他让线程2先使用，同样地，线程2也在谦让，致使双方还是无法进行。

另外，如果对同样一个资源，多次调用非递归锁，会造成多重锁死；

（3）性能问题-主要是线程切换，不作为本文重点。

五、C++11中的多线程

C++11将Boost库中的thread类引进——std::thread,同时提供了std::atomic,std::mutex,std::conditonal_variable,std::future,配上智能指针，使得c++11中的多线程并发变得安全容易。

C++11 所定义的线程是和操作系的线程是一一对应的，也就是说我们生成的线程最终还是直接接受操作系统的调度的。

5.1、创建和结束线程

创建一个新线程，实质就是定义一个thread类，而对该类的初始化的不同对应不同的几种方式：

（1）使用一个函数指针作为入口，传入相关参数

shared_ptr<boost::thread> thread_;

thread_.reset(new thread(funcReturnInt, "%d%s\n", 100, "\%"));

（2）使用C++11中的匿名函数：

thread_.reset(new thread( [](int ia, int ib){
cout << (ia + ib) << endl;},a,b ));

（3）默认构造，创建空的线程：

thread_.reset(new thread());

（4）C++新特性，移动构造，移动构造后t1就析构了：

std::thread t2(std::move(t1))

同时，该类的复制构造函数：thread (const thread&) = delete;意味着该类对象不能被复制构造。

以下是四种构造方式的实例：

#include <iostream>
#include <utility>
#include <thread>//std::thread,std::this_thread
#include <functional>
#include <atomic>

void f1(int n)
{
for (int i = 0; i < 5; ++i) {
std::cout << "Thread "
<< std::this_thread::get_id()
<< " executing\n";
}
}

int main()
{
int n = 0;
std::thread t1; // 空线程，实际上也什么都没有。
std::thread t2(f1, n);
std::thread t3( [](int ia, int ib){
cout << (ia + ib) << endl;},a,b );
std::thread t4(std::move(t2))//移动构造之后，t2已经析构掉了
t3.join();
t4.join();
std::cout << "Final value of n is " << n << '\n';
}

结束一个进程，之前在CAFFE源码中讲过有两种方式，一是等待线程自己结束，调用join()等待目标线程结束为止；二是将线程分离dispatch()，再主动杀死线程，但是C++11不能直接结束线程，所以只能被动等待。

5.2、线程调度

C++11中没有调度策略有关的类或者函数

5.3、data_racing

5.3.1 atomic

对于基本数据类型，可以采用原子访问：

原本需要三步的自增操作，现在必须一口气完成，中间不能中断，也就出现不了线程干扰的问题。

atomic内容很多，以后再单独学习。

atomic<int> a(0) ;
thread ta( func_inc, &a);

5.3.2 std::mutex

Mutex 又称互斥量,提供了独占所有权的特性。

mutex有四类：

std::mutex//最基本的 Mutex 类。
std::recursive_mutex//递归 Mutex 类。
std::time_mutex//定时 Mutex 类。
std::recursive_timed_mutex//定时递归 Mutex 类。


当我们可能会在某个线程内重复调用某个锁的加锁动作时，我们应该使用递归锁 (recursive lock)，除此之外，统统使用最基本的mutex即可。

std::mutex不允许拷贝构造，也不允许 move 拷贝，最初产生的 mutex 对象是处于 unlocked 状态的。

锁有两类：

i、使用std:: lock_guard，lock_guard 是一个范围锁，本质是 RAII(Resource Acquire Is Initialization)，在构建的时候自动加锁，在析构的时候自动解锁，这保证了每一次加锁都会得到解锁，但这并不意味着lock_guard 对象能决定 Mutex 对象的生命周期，只负责锁。

#include <iostream>
#include <vector>
#include <thread>
#include <mutex>
std::mutex mtx

void print_hello(int label){
std::lock_guard<std::mutex> lck (mtx);
std::cout<<"hello thread"<<label<<'\n';
}

int main () {
std::vector<std::thread> threads;
for (int i=0; i<10; ++i)
threads.push_back(std::thread(print_hello,i+1));

for (auto& th : threads) th.join();
}


ii、手动加锁：std::unique_lock 或者直接使用 mutex，mutex中调用lock、unlock等：

std::unique_lock 同样是基于RAII，负责自动加锁和解锁，但是他提供了手动上锁的机会。

std::unique_lock 的构造函数相当多，提供了多种选择。

其函数方法除了加锁和解锁：

lock，

try_lock，如果mutex已经被其他线程锁住某，上锁失败，返回false。

try_lock_for，try_lock_until 和 unlock

还有swap()：与另一个 std::unique_lock 对象交换它们所管理的 Mutex 对象的所有权;

mutex():返回指向mutex对象指针;

release():返回指向mutex对象指针，释放所有权

owns_lock（）：返回是否获得锁。

void print_thread_id (int id) {
std::unique_lock<std::mutex> lck (mtx,std::defer_lock);//此时并没有锁住mutex

lck.lock();//手动加锁
std::cout << "thread #" << id << '\n';
lck.unlock();//手动解锁
}


try_lock()返回布尔值，可以用于判断。

if (lck.try_lock())
std::cout <<"thread #" << id << '\n';
else
std::cout << 'mutex has been locked';


纯手工加锁

std::mutex mtx;
...
mtx.lock();
(*p)++;
mtx.unlock();
...


（3）std::condition_variable

条件变量：条件变量可以让一个线程等待其它线程的通知 (wait，wait_for，wait_until)，也可以给其它线程发送通知 (notify_one，notify_all)，条件变量必须和锁配合使用，在等待时因为有解锁和重新加锁，所以，在等待时必须使用可以手工解锁和加锁的锁，比如 unique_lock，而不能使用 lock_guard。

以下使用CAFFE中的例子：

完成push操作，向另一个线程发送通知：

template<typename T>
class BlockingQueue<T>::sync {
public:
mutable boost::mutex mutex_;#互斥锁
boost::condition_variable condition_; #条件变量
};

template<typename T>
void BlockingQueue<T>::push(const T& t) {
boost::mutex::scoped_lock lock(sync_->mutex_);
queue_.push(t);
lock.unlock();
sync_->condition_.notify_one();
}


接收到其他线程的通知，立即唤醒该线程,否则wait()会自动调用 lck.unlock() 释放锁,使得其他在等待资源的线程可以获得该锁：

template<typename T>
T BlockingQueue<T>::pop(const string& log_on_wait) {
boost::mutex::scoped_lock lock(sync_->mutex_);

while (queue_.empty()) {
if (!log_on_wait.empty()) {
LOG_EVERY_N(INFO, 1000)<< log_on_wait;
}
sync_->condition_.wait(lock);
}

T t = queue_.front();
queue_.pop();
return t;
}


（5）线程本地存储机制：对于共享资源，TSL保证每个线程拥有一个资源的副本，然后允许各线程访问各自对应的副本，最后再将副本合并。实现的机制就是建立一个查找表，根据线程的id读取线程对应的那一份数据。

例如CAFFE中：

static boost::thread_specific_ptr<Caffe> thread_instance_;


c++11中：

std::thread_specific_ptr<Caffe> thread_instance_


5.4、活跃度问题

（1）死锁：

死锁的条件[2]

资源互斥，某个资源在某一时刻只能被一个线程持有 (hold)；

吃着碗里的还看着锅里的，持有一个以上的互斥资源的线程在等待被其它进程持有的互斥资源；

不可抢占，只有在某互斥资源的持有线程释放了该资源之后，其它线程才能去持有该资源；

环形等待，有两个或者两个以上的线程各自持有某些互斥资源，并且各自在等待其它线程所持有的互斥资源。

（2）第二个问题，使用条件变量唤醒线程，在更高层次的队列中还可以使用对偶模型进行唤醒。

（3）当我们可能会在某个线程内重复调用某个锁的加锁动作时，我们应该使用递归锁 (recursive lock)，在 C++11 中，可以根据需要来使用 recursive_mutex，或者 recursive_timed_mutex。

参考资料：

[1]进程与线程的一个简单解释

[2]使用c++11编写多线程程序