C++11多线程基本使用

C++11增加了线程及线程相关的累,很方便的支持了并发编程,使得编写的多线程程序的可移植性得到了很大的提高.

线程的创建

用std::thread 创建线程非常的简单,只需要提供线程函数或者函数对象即可,并可以同时指定线程的参数:

#include<iostream>
#include<thread>
#include<chrono>
using namespace std;

//线程函数
void func(int a, int b, int c)
{
std::this_thread::sleep_for(std::chrono::seconds(3));
cout << a << " " << b << " " << c << endl;
}

int main()
{
//创建线程对象t1,绑定线程函数为func
std::thread t1(func, 1, 2, 3);
//输出t1的线程ID
std::cout << "ID:" << t1.get_id() << std::endl;
//等待t1线程函数执行结束
t1.join();
std::thread t2(func, 2, 3, 4);
//后台执行t2的线程函数,并且不会因为main函数结束时,线程函数未执行完而产生异常
t2.detach();
cout << "after t2 ,main is runing" << endl;
//以lambda表达式作为被帮顶的线程函数
std::thread t4([](int a, int b, int c)
{
//线程休眠5秒
std::this_thread::sleep_for(std::chrono::seconds(5));
cout << a << " " << b << " " << c << endl;
}, 4,5,6);
t4.join();

//获取CPU的核数
cout << "CPU: " << thread::hardware_concurrency() << endl;
//当添加下面注释掉的语句会抛出异常,因为线程对象先于线程函数结束了,应该保证线程对象的生命周期在线程函数执行完时仍然存在.
//std::thread t3(func, 3, 4, 5);

return 0;
}

线程函数将会运行于线程对象t中,join函数将会阻塞线程,直到线程函数执行结束,如果线程函数有返回值,返回值将被忽略.

detach可以将线程与线程对象分离,让线程作为后台线程执行,当前线程也不会阻塞了.但是detach之后就无法在和线程发生联系了.如果线程执行函数使用了临时变量可能会出现问,线程调用了detach在后台运行,临时变量可能已经销毁,那么线程会访问已经被销毁的变量.join能保证.

虽然这种方式创建线程很方便,但是std::thread 出了作用域后将会析构,这个时候线程函数还没执行完则会发生错误.

线程不可以复制但是可以移动.但是线程移动后,线程对象将不再代表任何线程了:

std::thread t(func);
//移动后,线程对象t不在代表任何线程
std::thread t1(std::move(t));
// t.join();
t1.join();

互斥量

互斥量是一种同步原语,是一种线程同步的手段,用来保护多线程同时访问的共享数据.

std::mutex: 独占的互斥量,不能递归使用.

std::timed_mutex: 带超时的独占互斥量,不能递归使用.

std::recursive_mutex: 递归互斥量,不带超时功能.

std::recursive_timed_mutex: 带超时的递归互斥量.

这些互斥量的基本接口十分相近,都是通过lock()来阻塞线程,直到获得互斥量的所有权为止.在线程或的互斥量并完成任务后,就必须使用unlock()来解除对互斥量的占用,lock和unlock必须成对出现.try_lock()尝试锁定互斥量,成功返回true,失败返回false,他是非阻塞的.

#include<iostream>
#include<thread>
#include<mutex>
using namespace std;

std::mutex g_lock;

void func()
{
//上锁
g_lock.lock();
cout << "in id: " << this_thread::get_id() << endl;
this_thread::sleep_for(chrono::seconds(1));
cout << "out id: " << this_thread::get_id() << endl;
//解锁
g_lock.unlock();
}

void f()
{
//lock_guard在构造时会自动锁定互斥量,而在退出作用域后进行析构时就会自动解锁.
lock_guard<std::mutex> lock(g_lock);
cout << "in id: " << this_thread::get_id() << endl;
this_thread::sleep_for(chrono::seconds(1));
cout << "out id: " << this_thread::get_id() << endl;
}

int main()
{
std::thread t1(func);
std::thread t2(func);
std::thread t3(func);

t1.join();
t2.join();
t3.join();

std::thread t4(f);
std::thread t5(f);
std::thread t6(f);

t4.join();
t5.join();
t6.join();
}

lock_guard用到了RAII的技术,这种技术在类的构造函数中分配资源,在析构函数中释放资源,保证资源在出了作用域之后就释放.

std::recursive_mutex递归锁允许同一个线程多次获得互斥量.但是尽量不要使用递归锁:

需要用到递归锁定的多线程互斥处理往往本身就是可以简化的,允许递归互很容易放纵复杂逻辑的产生,从而导致一些多线程同步引起的晦涩问题.

递归锁比起非递归锁,效率会低.

递归锁虽然允许同一个线程多次获得同一互斥量,但是可重复获得的最大次数并未具体说明,一旦超过一定次数就会抛出异常.

带超时的互斥量在获取锁的时候增加了超时等待功能,因为有时不知道获取锁需要多久,为了不至于一直等待获取互斥量,就设置一个等待超时时间,在超时后还可以做其他的的事情.

#include<iostream>
#include<thread>
#include<mutex>
#include<chrono>
using namespace std;

std::timed_mutex mutex1;

void work()
{
//设置阻塞时间
std::chrono::milliseconds timeout(100);

while (true) {
//带超时的锁,当阻塞超过100milliseconds时返回false
if (mutex1.try_lock_for(timeout)) {
cout << this_thread::get_id() << ": do work with the mutex" << endl;
std::chrono::milliseconds sleepDuration(250);
this_thread::sleep_for(sleepDuration);
} else {
cout << this_thread::get_id() << ": do work without mutex" << endl;

chrono::milliseconds sleepDuration(100);
std::this_thread::sleep_for(sleepDuration);
}
}
}

int main()
{
std::thread t1(work);
std::thread t2(work);

t1.join();
t2.join();

return 0;
}

条件变量

条件变量阻塞一个或多个线程，直到收到另外一个线程发来的通知或者超时，才会唤醒当前阻塞的进程,条件变量需要和互斥量配合使用.

C++11提供了两种条件变量

std::condition_variable，配合std::unique_lock进行wait操作

std::condition_variable_any，和任意带有lock，unlock的mutex进行搭配使用，比较灵活但效率略低。

条件变量的使用过程如下:

拥有条件变量的线程获取互斥锁

循环检查某个条件,如果条件不满足,则阻塞直到条件满足,如果条件满足,则向下执行.

某个线程满足条件执行完之后调用notify_one或notify_all唤醒一个或者所有的等待线程.

eg:

//同步队列的实现
#include<iostream>
#include<thread>
#include<mutex>
#include<condition_variable>
#include<list>
using namespace std;

template <typename T>
class SyncQueue
{
private:
//数据缓冲区
std::list<T> m_queue;
//互斥锁
std::mutex m_mutex;
//不为满的条件变量
std::condition_variable_any m_notFull;
//不为空的条件变量
std::condition_variable_any m_notEmpty;
//缓冲区最大大小
int m_maxsize;

//判断是否为满,因为给内部成员函数使用,而这些函数在调用前都已经上过锁了,所以无需在加锁
bool IsFull()
{
return m_queue.size() == m_maxsize;
}

//判断是否为空
bool IsEmpty()
{
return m_queue.empty();
}
public:
SyncQueue(int max):m_maxsize(max) {  }
//相缓冲区添加数据
void Put(const T& x)
{
//unique_lock与lock_guard相似,但是后者只能在析构时才释放锁,而前者可以随时释放锁
std::unique_guard<std::mutex> locker(m_mutex);

//若为满则需等待,而不能相缓冲区中添加
while (IsFull())
{
std::cout << "data Full" << std::endl;
//若为满,信号变量进行阻塞等待,此时释放m_mutex锁,然后直到被notify_one或者notify_all唤醒后先获取m_mutex锁
m_notFull.wait(m_mutex);
}
//相缓冲区添加数据
m_queue.push_back(x);
//唤醒处于等待中的非空条件变量
m_notEmpty.notify_one();
}

//从缓冲区获取数据
void Take(T& x)
{
std:unique_guard<std::mutex> locker(m_mutex);

//直接使用这种方法,就无需在定义私有的Empty,也无需写while循环判断了.
//m_notEmpty.wait(locker, [this] {return !m_queue.empty();});

//若为空则需等待,而不能从缓冲区中取出
while(IsEmpty())
{
std::cout << "data Empty" << std::endl;
m_notEmpty.wait(m_mutex);
}
//获取数据
x = m_queue.front();
//删除被获取的数据
m_queue.pop_front();
m_notFull.notify_one();
}

bool Empty()
{
std::lock_guard<std::mutex> locker(m_mutex);
return m_queue.empty();
}

bool Full()
{
std::lock_guard<std::mutex> locker(m_mutex);
return m_queue.size() == m_maxsize;
}

size_t Size()
{
std::lock_guard<std::mutex> locker(m_mutex);
return m_queue.size();
}
};

原子变量

C++11提供了一个原子类型

std::atomic<T>

,可以使用任意类型作为模板参数,C++11内置了整性的原子变量,使用原子变量就不需要使用互斥量来保护改变量了.

#include<atomic>

struct AtomicCounter {
std::atomic<int> value;

void increment()
{
++ value;
}

void decrement()
{
-- value;
}

int get()
{
return value;
}
};

call_once/once_flag

为了保证在多线程环境中某个函数仅被调用一次,例如,需要初始化某个对象,而这个对象智能被初始化一次的话,就可以使用std::call_once来保证函数在多线程环境下只调用一次.

#include<iostream>
#include<trhead>
#include<mutex>

std:once_flag flag;

void do_once()
{
std::call_once(flag, []() {std::cout << "called" << std::endl;});
}

int main()
{
std::thread t1(do_once);
std::thread t2(do_once);
std::thread t3(do_once);

t1.join();
t2.join();
t3.join();

return 0;
}