基本概念
- 并发：让实际可能串行发生的事情好像同时发生一样。
- 并行：并发序列同时执行，真正的并行只能在多核系统中存在。并行要求程序能够同时执行多个操作，而并发只要求程序能够假装同时执行多个操作((在单核下，通过os的调度实现并发，多核下会分配到不同的核去处理))
用一个典型的例子来说明并发与并行：100个人排队去银行窗口办理存款业务，那么基本流程是：1.客户到窗口排队–>2.轮到后填写业务单–>3.窗口业务员将业务单信息输入电脑–>4.业务员等待电脑处理完成–>5.通知下一个客户。它是一个串行的流程，如果我们要使用并发，则在业务员等待电脑处理的时间，可以让下一个客户先填业务单。也就是说第2步与第4可以并发处理，但是还是只有一个窗口服务这100个人。那么并行就是多开几个窗口同时服务这个100人。
- 支持并发的系统必须提供基本的核心功能:线程与同步原语。同步元语包括:互斥，条件变量，原子操作等。
那么并发编程就是通过线程，互斥，条件变量，原子操作等同步原语来保证共享数据的安全从而通过线程间的协作来提高程序效率及性能一种程序设计方法。
多线程编程的层次
- 理解同步原语的语义及使用场景，比如互斥锁，条件变量，原子操作的语义及使用场景。以及并发编程基本组件的使用场景，比如消费者生产者队列，异步消息，线程池等。
- 上面提到的同步原语的语义在不同操作系统中是相同的，但是所提供支持的 API肯定是不同的。第二个层面就是掌握操作系统中这些API的用法和掌握如何通过这些API去实现并发编程的基本组件。
- 掌握多线程的设计思路，程序的哪些功能可以使用线程，线程间如何同步，如何交互。这个层面属于设计思想的范畴，是综合运用。
并发编程中第1，3才是学习并发编程的核心所在，这些基础语义，组件及设计思想，是跨系统，跨语言的。，至于第2点有了线程库的支持后，就显得的无关紧要了。比如在java中，编写多线程程序，肯定是直接使用标准库中的thread，不会去直接面对os的API，当然thread库的实现还是依赖os的API，所以C++11中既然有了thread库，那么我们应该学会使用它，而不再去直接API或是使用第三方库。
线程库
操作系统提供了很多系统级别的API支持多线程编程，但是这样的API众多
- 比如windows下提供的创建线程的API有_beginthread,CreateThread,互斥的API：WaitForSingleObject,WaitForMultipleObjects,临界区系统API等。
- linux下posix -pthreads又是一套与windows平台完全不同的API体系。
不同系统中的并发编程的基本要素的语义都是相同的，只是API的不同。所以为了简化跨平台的差异性，就出现了很多对API进行封装的线程库。在C++ 98 STL中并没有线程库，往往使用的是第三方库，那么从C++ 11开始，STL中就已经提供了线程库，提供了线程，互斥量，条件变量，原子操作的封装。
STL中thread库
线程的创建
C++ 11创建一个线程，变得很简单
#include <thread>
#include <iostream>
void func()
{
std::cout<<"func()"<<std::endl;
}
int main()
{
std::thead t(func);
//等待线程执行完毕
t.join();
//线程分离
//t.detach();
}
std::thread是线程类，构造一个线程对象即产生一个运行的线程。构造函数中传入一个可调用对象作为线程的执行体，可调用对象包括，函数，函数指针，函数对象，lambda表达式等。在上面的例子中，线程的执行体就是一个普通函数。
向线程执行体传入参数
#include <thread>
#include <iostream>
void func(int i)
{
std::cout<<"i:"<<i<<std::endl;
}
int main()
{
std::thread t(func,18);
t.join();
}
注意这里实参是通过copy或移动存入执行体可调用对象的，可以通过std::ref转换成引用传入。对于一个不支持拷贝对象或是拷贝开销很重的对象，通过std::ref转换传入。
将对象的成员函数作为线程的执行体
#include <iostream>
#include <thread>
class CTest
{
public:
CTest(){}
~CTest(){}
public:
void func()
{
std::cout<<"this CTest:func()"<<std::endl;
}
};
int main()
{
CTest t;
std::thread t1(&CTest::func,&t);
t1.join();
}
类的成员函数，第一个参数是所属对象的指针，所以上面的代码传入&t
当然也可以用std::bind产生一个新的可调用对象(执行体)传入thread构造函数
std::thread t1(std::bind(&CTest::func,&t));
线程的生命周期
线程的生命周期等于线程对象的生命周期，当线程对象被销毁时，如果线程还未结束。此时，程序会异常终止，如下代码：
#include <thread>
#include <chrono>
void func()
{
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout<<"thread end "<<std::endl;
}
int main()
{
std::thread t1(func);
}
主线线程执行完时，t1对象被析构，而它代表的线程在sleep中并未结束，如果主线程中没有调用thread的detach方法(放弃对线程的管理),如上代码此时程序会中止，main end 语句不会输出。所以要保证线程对象的生命周期大于线程周期，或者调用detach或join方法。
线程对象不支持复制和赋值，只支持移动语义
#include <thread>
#include <chrono>
#include <iostream>
void func()
{
std::cout << "before sleep,thread id " << std::this_thread::get_id() << std::endl;
std::this_thread::sleep_for(std::chrono::seconds(3));
std::cout << "after sleep,thread id " << std::this_thread::get_id() << std::endl;
}
int main()
{
std::thread t1 = std::thread(func);
std::this_thread::sleep_for(std::chrono::seconds(1));
//std::thread t2(t1) 编译不过
std::thread t2(std::move(t1));
t2.join();
}
线程的复制构造函数，赋值函数都被标示为delete
将线程移动到容器中
#include <thread>
#include <vector>
#include <iostream>
void func()
{
std::cout << "this is thread " << std::this_thread::get_id();
std::this_thread::sleep_for(std::chrono::seconds(1));
}
int main()
{
std::thread t(func);
std::vector<std::thread> vecThreads;
for (int i = 0; i < 3; ++i)
{
//push_back(t) 会编译不过，因为这里调用的复制版本的push_back
//这里调用的是push_back移动版本
vecThreads.push_back(std::move(t));
}
}
线程间的通信
线程间通信的两个问题：
临界区
把对共享资源进行访问的程序片段称为临界区，可以通过互斥量和条件变量来保护临界区，互斥量和条件变量是OS的同步原语
互斥量
互斥量有两种状态，解锁和加锁，先看看关于临界区互斥的例子：
#include <iostream>
#include <thread>
int g_counter = 0;
void func(int cnt)
{
for (int i = 0; i < cnt; ++i)
{
g_counter++;
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
}
int main()
{
std::thread t1(func, 100);
std::thread t2(func, 100);
t1.join();
t2.join();
std::cout << "counter " << g_counter << std::endl;
}
上面的代码两个线程t1,t2都会访问，修改共享资源 g_counter，分别对它递增100。最后预期的结果应该是200，但是会出现非200的情况。
临界区代码g_counter++ 是非原子操作，会被编译成多条汇编语句，在OS对线程进行调度时，比如t1线程当前读到的值是10，此时调度t2线程执行，读到值也是10，两个线程分别作自增操作，值变为了11，尽管都做了自增操作。
可以通过C++11 中std::mutex的lock和unlock对临界区进行保护
#include <iostream>
#include <thread>
int g_counter = 0;
std::mutex m;
void func(int cnt)
{
for (int i = 0; i < cnt; ++i)
{
m.lock();
g_counter++;
m.unlock();
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
}
int main()
{
std::thread t1(func, 100);
std::thread t2(func, 100);
t1.join();
t2.join();
std::cout << "counter " << g_counter << std::endl;
}
上面的代码，直接通过mutex的lock和unlock方法，保护临界区。可以叫做以手动的方式来加/解锁，但是这种mutex的使用方式容易造成死锁，如果在unlock之前，代码因为某些原因而没有调用unlock，此时就会造成死锁。
C++ 11中提供了std::lock_guard,std::unique_lock,它们通过RAII方法实现对锁的自动释放(超出作用域，lock对象被析构时，会被自动释放)。那么保护临界区的代码如下
int g_int = 0;
std::mutex g_mutex;
void func()
{
{
//在超出作用域时，释放锁
std::lock_gruad<std::mutex> lock(g_mutex);
g_int = 18;
}
}
-
std::lock_guard相较 std::unique_lock更简单，纯粹些。完全是通过RAII来封装std::mutex，在构造时加锁，在析构时解锁，不提高额外的方法。
-
std::unique_lock，还提供了手动控制锁的方法:lock()与unlock(),相比std::lock_guard它提供了更多lock的策略，比如try_lock，try_lock_until等。它更多的是配合条件变量来使用。
C++ 11提供了4种语义的互斥量:独占互斥量 std::mutex,带超时的独占互斥量 std::timed_mutex,递归互斥量 std::recursive_mutex，带超时的递归互斥量 std::recursive_timed_mutex。
在实际的应用中，应该优先使用std::mutex，它满足所有的使用互斥量的场景。
条件变量
条件变量也用来保护临界区，用于线程间的协作。当线程工作的前提必须是共享资源满足某个条件时，通常会使用条件变量来实现(就是线程间的相互通知，线程A操作完共享资源后，通知线程B)
C++ 11中的条件变量condition_variable，需要配合unique_lock使用先看一个例子：
#include <thread>
#include <condition_variable>
#include <iostream>
std::mutex g_mtx;
//条件变量
std::condition_variable g_cnd;
//共享资源
int g_cnt = 0;
void thread1()
{
std::unique_lock<std::mutex> lock(g_mtx);
g_cnd.wait(lock);
std::cout << "g_cnt " << g_cnt << std::endl;
}
int main()
{
std::thread t(thread1);
std::this_thread::sleep_for(std::chrono::seconds(1));
while(1)
{
//std::lock_guard<std::mutex> lock(g_mtx);
//为了缩小锁的范围，调用mutex的lock和unlock进行手动加解锁
g_mtx.lock();
++ g_cnt;
if (18 == g_cnt)
{
g_mtx.unlock();
//cnd通知时不必加锁
g_cnd.notify_all();
}
if (118 == g_cnt)
{
g_mtx.unlock();
break;
}
g_mtx.unlock();
}
std::cout<<"break "<<std::endl;
t.join();
}
上面例子产生两个线程，分别是主线程和thread1，在thread1中会等待条件变量g_cnd，在主线程中当g_cnt为18时，会调用g_cnd.notify_all();来唤醒阻塞在g_cnd.wait()上的thread1，thread1被唤醒后会再次获得g_mtx锁，这里有几个注意的点：
-
是否输出g_cnd的值依赖的是thread1线程和主线程执行的时序，如果主线程先执行。thread1会一直阻塞在g_cnd.wait()上，所以在主线程中添加语句 std::this_thread::sleep_for(std::chrono::seconds(1));的目的是让thread1先执行，先阻塞到g_cnd.wait()上，能等到g_cnd的notify -
条件变量的notify系列方法是不用加锁的。所以为了缩小加锁范围，调用的是std::mutexlock和unlock方法，手动加解锁。在这个例子中，因为是一个读线程，一个写线程分别读写g_cnt，也可以不用加锁
错过信号及假醒
上面的例子中，存在thread1线程错过g_cndnotify的情况，这种情况是属于错过信号。
假醒的一个例子
#include <thread>
#include <condition_variable>
#include <mutex>
#include <iostream>
std::mutex g_mxt;
//条件变量
std::condition_variable g_cnd;
//共享资源
int g_cnt = 0;
bool g_flag = false;
void thread1Func()
{
std::unique_lock<std::mutex> lock(g_mxt);
g_cnd.wait(lock);
if (g_flag)
{
std::cout << "thread_1 g_cnt : " << g_cnt << std::endl;
g_flag = false;
}
else
{
std::cout << "thread_1 g_flag is false" << std::endl;
}
}
void thread2Func()
{
std::unique_lock<std::mutex> lock(g_mxt);
g_cnd.wait(lock);
if (g_flag)
{
std::cout << "thread_2 g_cnt : " << g_cnt << std::endl;
g_flag = false;
}
else
{
std::cout << "thread_2 g_flag is false " << std::endl;
}
}
int main()
{
std::thread thread1(thread1Func);
std::thread thread2(thread2Func);
std::this_thread::sleep_for(std::chrono::seconds(1));
while (1)
{
++g_cnt;
if (18 == g_cnt)
{
g_flag = true;
g_cnd.notify_all();
break;
}
}
thread1.join();
thread2.join();
}
上面的代码简单的模拟了一个假醒的场景，g_flag作为共享资源会被两个线程thread1,thread2竞争，在g_cnt为18时，也会通过notify_all通知两个线程，但是两个线程中只会有一个线程会得g_flag。那么另外一个线程的被唤醒就是假醒(在真正的业务场景，线程可能会拿到共享资源后才会去执行具体的业务。这里的示例代码只是简单的等待一个g_flag再输出)。
通常条件变量被唤醒，需要配合一个判断条件。当这个条件不满足时，线程应该会再次阻塞。如果条件满足则不会被阻塞。，形式入下：
while(!condition)
{
g_cnd.wait(g_mtx);
}
还可以解决假醒的情况，假醒就是虚假唤醒，当线程被条件变量唤醒时，但此时共享资源并不满足线程继续执行的条件，比如
多个线程都对一个队列取数据，当队列中没用数据时，线程都被阻塞，当有数据时，线程都被唤醒去数据。但是可能存在某个线程醒来，但是队列为空的情况。出现这种情况，是某个线程先把数据取走了。也就是
无论哪一个线程抢到了资源, 另一个线程的唤醒就可以被认为是没有必要的, 也就是被虚假唤醒了。
虚假唤醒有应用层的业务实现造成，这种情况下是跟线程调度时序相关。也有跟操作系统对条件变量的实现相关，也会产生虚假唤醒的情况，机制比较复杂。但是还是要意识到存在虚假唤醒的情况。
condition_variable 有一个重载的wait方法
template< class Predicate >
void wait( std::unique_lock<std::mutex>& lock, Predicate pred )
第二个Predicate就是一个判断式，达到的效果如同以下语句
while (!pred()) {
wait(lock);
}
当 pred()返回flase时，再调用wait方法。
改动之后的代码如下：
#include <thread>
#include <condition_variable>
#include <iostream>
std::mutex g_mtx;
//条件变量
std::condition_variable g_cnd;
//共享资源
int g_cnt = 0;
//flag
bool g_flag = false;
void thread1()
{
std::unique_lock<std::mutex> lock(g_mtx);
g_cnd.wait(lock,[]{return g_flag;});
std::cout << "g_cnt " << g_cnt << std::endl;
}
int main()
{
std::thread t(thread1);
std::this_thread::sleep_for(std::chrono::seconds(1));
while(1)
{
//std::lock_guard<std::mutex> lock(g_mtx);
//为了缩小锁的范围，调用mutex的lock和unlock进行手动加解锁
g_mtx.lock();
++ g_cnt;
if (18 == g_cnt)
{
g_mtx.unlock();
g_flag = true;
//cnd通知时不必加锁
g_cnd.notify_all();
}
if (118 == g_cnt)
{
g_mtx.unlock();
break;
}
g_mtx.unlock();
}
t.join();
}
上面的代码额外定义了判断条件g_flag，wait方法的 Predicate 传入的一个lambda表达式，当g_flag为false时，线程阻塞到条件变量上。
条件变量的使用规则
总结条件变量的用法：
它允许线程由于一些未到的条件而阻塞。它通常与互斥量结合在一起使用。这种方式用于让一个线程锁住一个互斥量，然后当它不能获得它期待的结果时等待一个条件变量(并释放该锁)，最后另一个线程会向它发信号，使它可以继
续执行(获取锁)
- 获得锁(lock_guard)
- 在保有锁时对共享资源进行修改
- 释放锁，通知其它线程(std::condition_variable 上执行 notify_one 或 notify_all)
- 任何有意在条件变量(std::condition_variable) 上等待的线程必须
- 获得锁(unique_lock) ，在与用于保护共享资源者相同的互斥锁上
- 执行等待(condition_variable的wait方法)，等待操作自动释放互斥，并悬挂线程的执行。
- 被通知时，时限消失或虚假唤醒发生，线程被唤醒，且自动重获得互斥。之后线程应检查条件，若唤醒是虚假的，则继续等待(condition_variable的wait方法)。
条件变量通常会配合一个检查条件使用，循环检查该条件，以防止"假醒"。通过wait方法配合lambda表达式可以很方便的实现(形参是一个调用对象，不限于lambda表达式)。
|