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递归锁允许同一个线程多次获得互斥量.但是尽量不要使用递归锁:

  1. 需要用到递归锁定的多线程互斥处理往往本身就是可以简化的,允许递归互很容易放纵复杂逻辑的产生,从而导致一些多线程同步引起的晦涩问题.
  2. 递归锁比起非递归锁,效率会低.
  3. 递归锁虽然允许同一个线程多次获得同一互斥量,但是可重复获得的最大次数并未具体说明,一旦超过一定次数就会抛出异常.

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

#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提供了两种条件变量

  1. std::condition_variable，配合std::unique_lock进行wait操作
  2. std::condition_variable_any，和任意带有lock，unlock的mutex进行搭配使用，比较灵活但效率略低。

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

  1. 拥有条件变量的线程获取互斥锁
  2. 循环检查某个条件,如果条件不满足,则阻塞直到条件满足,如果条件满足,则向下执行.
  3. 某个线程满足条件执行完之后调用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;
}