C++ 多线程基础及 C++11 多线程库总结

墨蓝 2022-12-06 15:51 214阅读 0赞

# 1. 多线程基础 #

## 1.1 进程与线程 ##

*  根本区别：  
    进程是操作系统资源分配的基本单位，线程是任务调度和执行的基本单位
 *  开销方面：  
    每个进程都有自己独立的代码和数据空间，程序之间的切换开销较大。  
    线程可以看作是轻量级的进程，同一类线程共享代码和数据空间，每个线程都有自己独立的运行栈和程序计数器，线程之间切换开销小。
 *  所处环境：  
    一个操作系统能同时运行多个进程（程序）。  
    在一个进程中，可以有多个线程同时执行。
 *  内存分配方面：  
    系统在运行的时候会为每个进程分配不同的内存空间。  
    对线程而言，系统不会为线程分配内存（线程使用的资源，来自于其所属进程的资源），线程组之间只能共享资源。
 *  包含关系：  
    没有线程的进程可以看作是单线程。一个进程可以包含多个线程，每个进程有且只有一个主线程  
    线程是进程的一部分，所以线程也称为轻量级进程-

## 1.2 并发与并行 ##

如果某个系统支持两个或者多个动作（Action）同时存在，那么这个系统就是一个并发系统。如果某个系统支持两个或者多个动作同时执行，那么这个系统就是一个并行系统。

你吃饭吃到一半，电话来了，你一直到吃完了以后才去接，这就说明你不支持并发也不支持并行。
    你吃饭吃到一半，电话来了，你停了下来接了电话，接完后继续吃饭，这说明你支持并发。
    你吃饭吃到一半，电话来了，你一边打电话一边吃饭，这说明你支持并行。

并行”概念是“并发”概念的一个子集。也就是说，你可以编写一个拥有多个线程或者进程的并发程序，但如果没有多核处理器来执行这个程序，那么就不能以并行方式来运行代码。因此，凡是在求解单个问题时涉及多个执行流程的编程模式或者执行行为，都属于并发编程的范畴。

并发是不是一个线程，并行是多个线程？  
答：并发和并行都可以是多个线程，就看这些线程能不能同时被（多个）cpu(物理线程)执行，如果可以就是并行，而并发誓多个线程被 cpu 轮流切换着执行。

*  进程并发
 *  线程并发

## 1.3 进程通信 ##

*  同一台PC：管道，文件，消息队列，共享内存
 *  不同PC：socket

## 1.4 线程通讯 ##

*  锁机制：包括互斥锁、条件变量、读写锁  
    互斥锁提供了以排他方式防止数据结构被并发修改的方法。  
    读写锁允许多个线程同时读共享数据，而对写操作是互斥的。  
    条件变量可以以原子的方式阻塞进程，直到某个特定条件为真为止。对条件的测试是在互斥锁的保护下进行的。条件变量始终与互斥锁一起使用。
 *  信号量机制(Semaphore)  
    包括无名线程信号量和命名线程信号量。
 *  信号机制(Signal)  
    类似进程间的信号处理。

线程间的通信目的主要是用于线程同步，所以线程没有像进程通信中的用于数据交换的通信机制。

## 1.5 线程并发 ##

*  一个进程中的所有线程共享地址空间，因此全局变量，指针，引用可以在线程之间传递
 *  数据一致性问题
 *  线程之间切换需要保存中间量，消耗资源，所以线程不是越多越好，极限差不多在 2000 个线程，或者根据硬件，比如 CPU\*2，或者根据业务需要，实际中一般不操作500个，控制在 200 个以内，否则效率太低。

# 2. C++ 11 多线程库 #

c++ 11 开始语言本身提供多线程支持，因此可以实现跨平台，可移植性。

## 2.1 创建线程 ##

*  \# include <thread>
 *  初始函数，函数结束，他的所有线程结束
 *  主线程结束，它的子线程结束

#include <iostream>
    #include <thread>
    
    using namespace std;
    
    // 入口函数
    void entry(int a)
    {
    	cout << "entry sub-thread" << endl;
    }
    
    int main()
    {
    	thread threadObj(entry， 6);    //入口函数，参数
        threadObj.join();   //等待线程结束
        // threadObj.detch();
    	cout << "entry main thread" << endl;
    	return 0;
    }

类的成员函数作为入口函数

class A {
    public:
    	A(int i) : m_i(i)
    	{
    		cout << "construct! thread ID:  " << std::this_thread::get_id() << endl;
    	}
    
    	void print(const A& a)
    	{
    		cout << "sub_thread ID: " << std::this_thread::get_id() << endl;
    	}
    	int m_i;
    };
    
    int main()
    {
    	int n = 1;
    	int& m = n;
    	A a(10);
    	cout << "main thread ID: " << std::this_thread::get_id() << endl;
    
    	thread mythread(&A::print, &a, a);  //传入成员函数地址、 类对象地址、参数
    	mythread.join();
        return 0;
    }

## 2.2 thread 的入口 ##

*  可调用对象（普通函数，类成员函数，类静态函数，仿函数，函数指针，重载了operate ()的类对象，lambda表达式，std::function）
 *  class 需要是可调用的类，即 void operator ()(), 注意以对象作为入口，对象会被复制到子线程中

## 2.3 thread 入口参数 ##

*  引用　vs实测背后发生了copy，所以无法通过引用来传值，需要注意的是，如果用引用需要同时用const，比如const A& a, 否则会报错，如果需要通过引用传值，需要用std::ref(或者加 & )

void print(const int& n)    //没有const会报错
    {
    	cout << "sub_thread: " << n << endl;
    }
    
    int main()
    {
    	int n = 1;
    	int& m = n;
    
    	thread mythread(print, m); //虽然print参数是引用，m 是引用但是会发生拷贝
        //thread mythread(print, std::ref(m)) //引用
    	mythread.join();
        return 0;
    }

*  指针　不安全，可能主线程已经销毁了内存，造成隐患，detach时一定会出问题
 *  临时参数（对象）可以帮助解决主线程退出的问题，即主线程退出之前会先构造好临时对象，具体来说，在创建线程时就构造临时对象，然后在线程入口函数里面用引用来接（否则会多一次拷贝构造）
 *  如果用隐式类型转换会有风险，因为隐式转换会在子线程中完成，如果detach的话，就会线程不安全
 *  如果参数是智能指针，如unique\_ptr, 需要用std::move(your unique\_ptr), 但是一定要用join，因为内存是共享的，否则会不安全
 *  成员函数指针

## 2.4 多个线程下保护共享数据 ##

### 2.4.1 mutex ###

#include "stdafx.h"
    #include <iostream>
    #include <thread>
    #include <mutex>
    #include <list>
    #include <vector>
    
    using namespace std;
    
    
    class Msg {
    public:
    	void InMsg()
    	{
    		for(int i = 0; i < 1000; ++i)
    		{
    			cout << "start input msg id = " << i << endl;
    			mut_Msg.lock();
    			m_Msg.push_back(i);
    			mut_Msg.unlock();
    			cout << "end input msg id = " << i << endl;
    		}
    	}
    
    	void OutMsg()
    	{
    		while (1)
    		{
    			if (bOutMsg())
    			{
    				cout << "pop out msg success!" << endl;
    			}
    			else
    			{
    				cout << "msg box is empty!" << endl;
    				_sleep(1000);
    			}
    		}
    	}
    
    	bool bOutMsg()
    	{
    		mut_Msg.lock();
    		if (!m_Msg.empty())
    		{
    			m_Msg.pop_front();
    			mut_Msg.unlock();
    			return 1;
    		}
    		else
    		{
    			mut_Msg.unlock();
    			return 0;
    		}
    	}
    
    private:
    	list<int> m_Msg;
    	mutex mut_Msg;
    };
    
    int main(void)
    {
    	Msg a;
    	thread thread1(&Msg::InMsg, &a);
    	thread thread2(&Msg::OutMsg, &a);
    	thread1.join();
    	thread2.join();
    	return 0;
    }

*  mutex 使用时应该尽量只保护需要保护的代码段。
 *  unlock 不能丢

### 2.4.2 lockguard ###

可以用 lockguard 接管 mutex,这样就不用手动　unlock, 传入　std::adopt\_lock　参数就是告诉　lockguard，锁已经锁了，只需要管理　unlock　就可以了。

lockguard 实际上是在其构造函数中调用了 lock(), 析构函数中调用 unlock().

std::lock(mutex1, mutex2);
    std::lockguard<std::mutex> guard1(mutex1, std::adopt_lock)
    std::lockguard<std::mutex> guard2(mutex2, std::adopt_lock)
    //....

lockguard 没有提供手动 lock & unlock 的接口。

### 2.4.3 死锁 ###

死锁，两个或以上的 lock 可能出现死锁。

threadA
    {
        mutexA.lock();
        mutexB.lock();
        //do some thing
        mutexA.unlock();
        mutexB.unlock();
    }
    
    threadB
    {
        mutexB.lock();
        mutexA.lock();
        //do some thing
        mutexA.unlock();
        mutexB.unlock();
    }

解决方法：

*  可以通过控制不同的线程lock的顺序来避免
 *  std::lock()　同时锁多个锁,如果有一个锁不上，它会 unlock 已经　lock 的锁

std::lock(mutex1, mutex2);
    //....
    mutex1.unlock();
    mutex2.unlock();

### 2.4.4 unique\_lock ###

unique\_lock 比 lockguard 更灵活，但是效率要低一些，占用内存更多。

unique\_lock 可以取代 lockguard，但是相比 lockguard，有更丰富的一些功能。

unique\_lock 支持以下参数：

*  adopt\_lock，意义和 lockguard 中一样表示已经lock了
 *  try\_to\_lock, 意味着 lock 和 unlock 都自动管理，可以判断是否lock成功，做不同操作,所以不会卡住。

std::unique_lock<std::mutex> guard1(mutex1, std::try_to_lock);
    if(guard1.owns_lock())  // get lock
    {
        //...
    }
    else
    {
        //...
    }

*  defer\_lock, 初始化一个没有加锁的 lock, 可以手动 lock，手动或者自动 unlock，  
    手动 lock 和 unlock 可以直接调用 unique\_lock 的成员函数：lock(), unlock()

unique\_lock 可以通过　release 来释放资源，即不再关联mutex。

unique\_lock 和 lock\_guard 都不能复制，但是unique\_lock 的所有权可以转移。

std::unique_lock<std::mutex> guard1(_mu);
    std::unique_lock<std::mutex> guard2 = guard1;  // error
    std::unique_lock<std::mutex> guard2 = std::move(guard1); // ok

### 2.5 线程安全的单例模式 ###

#include<iostream>
    #include<thread>
    #include<mutex>
    
    using namespace std;
    
    std::mutex instance_mutex;
    
    class SP {
    private:
    	SP() {}
    	static SP *m_pInstance; //static使得 m_pInstance 的作用域到程序结束
    	
    	class FREE {	//这个class专门负责 delete
    	public:
    		~FREE()
    		{
    			if (SP::m_pInstance)
    			{
    				delete SP::m_pInstance;
    				SP::m_pInstance = NULL;
    			}
    		}
    	};
    public:
    	static SP* GetInstance()	//static，否则无法直接调用
    	{
    		if (m_pInstance == NULL)	//双重锁定,提高运行效率，减少不必要的lock，unlock
    		{
    			std::unique_lock<std::mutex> mutex1(instance_mutex);	//c++ 11，自动lock，unlock
    			if (m_pInstance == NULL)
    			{
    				m_pInstance = new SP();
    				static FREE f;	//static 表示作用域直到程序推出，也就是说程序退出时会调用析构函数，从而达到自动释放内存的作用
    			}
    			return m_pInstance;
    		}
    	}
    	
    	static void Free()	//手动 delete
    	{
    		if (m_pInstance)
    		{
    			delete m_pInstance;
    			m_pInstance = NULL;
    		}
    	}
    };
    
    SP* SP::m_pInstance = NULL;
    
    
    int main(void)
    {
    	SP *p1 = SP::GetInstance();
    	SP *p2 = SP::GetInstance();
    	SP::Free();
    	SP::Free();
    	p1->Free();
    	p2->Free();
    }

### 2.6 call\_once() ###

std::call\_once() 是　c++ 11 引入的函数，保证某个函数只执行一次，具备互斥量的功能，但是比 mutex 高效, 适合比如 init 等场合

std::once_flag g_flag //决定 call_once 是否调用function
    
    std::call_once(g_flag, function_with_code_only_call_once)

## 2.7 condition\_variable ##

利用 condition\_variable 一般用来等待 unique\_lock, 可以提高程序执行效率。

condition\_variable 类有三个成员函数

*  wait() 等待一个条件成立
 *  notify\_one() 随机唤醒一个正在 wait 的线程，如果线程没有阻塞在 wait() 处，则没有办法唤醒
 *  notify\_all() 唤醒所有等待的线程

//*****thread A*****
    std::unique_lock<std::mutex> lock1(mutex1);
    
    //do some thing
    
    condition1.notify_one();
    //condition1.notify_all();
    
    //*****thread B*****
    std::unique_lock<std::mutex> lock1(mutex1);
    
    //do some thing
    
    condition1.wait(lock1); //如果没有第二个参数，wait将 release mutex1，然后阻塞，等待被唤醒
    //condition1.wait(lock1, [this]{ //第二个参数可以是任何可调用对象，如果表达式返回ture，直接return，如果为fasle，同上
        if(m_bStatus)
        {
            return true;
        }
        return false;
    })

*  虚假唤醒  
    线程被notify，但是却并没发执行，比如上面的例子，m\_bStatus 为false， 所以通过第二个参数可以防止虚假唤醒。

## 2.8 std::async, std::future ##

之前通过 thread 来创建线程，如果需要返回结果，可以通过全局变量/引用来实现，这里是另一种方式。

async 用于启动一个异步任务（创建线程并执行入口函数），返回一个 future 对象。通过 future 对象的 get() 获取入口函数返回的结果。

int entry()
    {
        //。。。
        return 1;
    }
    
    int main()
    {
        std::future<int> result = std::async(entry);    //entry 开始执行
        //std::future<int> result = std::async(std::launch::async, entry); //效果同上
        //std::future<int> result = std::async(std::launch::deferred, entry);   //延迟创建，等待 get/wait 才开始执行，如果没有调用，不会创建子线程
        int re = reult.get(); //get() 时会等待 entry 执行完， get() 不能调用多次
        //result.wait(); //不获取值返回值，等待线程
    }

## 2.9 std::packaged\_task ##

包装可调用对象，方便作为线程入口调用。

int entry(int a)
    {
        //。。。
        return a;
    }
    
    int main()
    {
        std::packaged_task<int(int)> pt(entry); //pt 本身就是一个可调用对象，类似函数，可以直接 pt(10),调用
        std::thread thread1(std::ref(pt), 1);
        thread1.join()
    
        std::future<int> result = pt.get_future(); //result 保存返回结果，可以 get
        //。。。
    }

## 2.10 std::promise ##

可以通过promise在线程直接传递值，一个线程往 promise 对象中写值，在其他线程中取值

void entry(std::promise<int> &prom, int a)
    {
        //。。。
        prom.set_value(result) 
        return;
    }
    
    void entry(std::future<int> & f)
    {
        //。。。
        result = f.get();
        return;
    }
    
    int main()
    {
        std::promise<int> prom;
        std::thread t1(entry, std::ref(prom), 10);
        t1.join()
    
        std::future<int> result = prom.get_future(); //result 保存返回结果,可以 get
    
        //。。。
    }

## 2.11 atomic ##

atomic 作用和 mutex 类似，不同点是 mutex 针对一个代码段，而 atomic 针对一个变量。  
atomic 操作相比 mutex 效率更高。

int g_count = 0;
    //*****mutex*****
    void entry()
    {
        mutex1.lock();
        g_count++;
        mutex1.unlock;
    }

std::atomic<int> g_count = 0;
    //*****atomic*****
    void entry()
    {
        g_count++;
    }

## 2.12 windows 临界区 ##

windows 临界区的概念和 mutex 类似。另外多次进入临界区是OK的，但是需要调用对应次数的出临界区。mutex 是不允许同一个线程中多次 lock 的。

include <windows.h>
    
    CRITICAL_SECTION winsec
    InitializeCriticalSection(winsec) //使用前必须初始化
    
    EnterCriticalSection(&winsec);
    EnterCriticalSection(&winsec);
    //do some thing
    LeaveCriticalSection(&winsec);
    LeaveCriticalSection(&winsec);

## 2.13 线程池 ##

#ifndef THREAD_POOL_H
    #define THREAD_POOL_H
    
    #include <vector>
    #include <queue>
    #include <memory>
    #include <thread>
    #include <mutex>
    #include <condition_variable>
    #include <future>
    #include <functional>
    #include <stdexcept>
    
    // 线程池类
    class ThreadPool {
    public:
    	// 构造函数，传入线程数
    	ThreadPool(size_t threads);
    	// 入队任务(传入函数和函数的参数)
    	template<class F, class... Args>
    	auto enqueue(F&& f, Args&&... args)
    		->std::future<typename std::result_of<F(Args...)>::type>;
    	// 一个最简单的函数包装模板可以这样写(C++11)适用于任何函数(变参、成员都可以)
    	// template<class F, class... Args>
    	// auto enqueue(F&& f, Args&&... args) -> decltype(declval<F>()(declval<Args>()...))
    	// {    return f(args...); }
    	// C++14更简单
    	// template<class F, class... Args>
    	// auto enqueue(F&& f, Args&&... args)
    	// {    return f(args...); }
    
    	// 析构
    	~ThreadPool();
    private:
    	// need to keep track of threads so we can join them
    	// 工作线程组
    	std::vector< std::thread > workers;
    	// 任务队列
    	std::queue< std::function<void()> > tasks;
    
    	// synchronization 异步
    	std::mutex queue_mutex;	// 队列互斥锁
    	std::condition_variable condition;	// 条件变量
    	bool stop;	// 停止标志
    };
    
    // the constructor just launches some amount of workers
    // 构造函数仅启动一些工作线程
    inline ThreadPool::ThreadPool(size_t threads)
    	: stop(false)
    {
    	for (size_t i = 0; i<threads; ++i)
    		// 添加线程到工作线程组
    		workers.emplace_back(	// 与push_back类型，但性能更好(与此类似的还有emplace/emlace_front)
    			[this]
    	{	// 线程内不断的从任务队列取任务执行
    		for (;;)
    		{
    			std::function<void()> task;
    
    			{
    				// 拿锁(独占所有权式)
    				std::unique_lock<std::mutex> lock(this->queue_mutex);
    				// 等待条件成立
    				this->condition.wait(lock,
    					[this] { return this->stop || !this->tasks.empty(); });
    				// 执行条件变量等待的时候，已经拿到了锁(即lock已经拿到锁，没有阻塞)
    				// 这里将会unlock释放锁，其他线程可以继续拿锁，但此处任然阻塞，等待条件成立
    				// 一旦收到其他线程notify_*唤醒，则再次lock，然后进行条件判断
    				// 当[return this->stop || !this->tasks.empty()]的结果为false将阻塞
    				// 条件为true时候解除阻塞。此时lock依然为锁住状态
    
    
    				// 如果线程池停止或者任务队列为空，结束返回
    				if (this->stop && this->tasks.empty()) {
    					return;
    				}
    				// 取得任务队首任务(注意此处的std::move)
    				task = std::move(this->tasks.front());
    				// 从队列移除
    				this->tasks.pop();
    			}
    			// 执行任务
    			task();
    		}
    	}
    	);
    }
    
    // add new work item to the pool
    // 添加一个新的工作任务到线程池
    template<class F, class... Args>
    auto ThreadPool::enqueue(F&& f, Args&&... args)
    -> std::future<typename std::result_of<F(Args...)>::type>
    {
    	using return_type = typename std::result_of<F(Args...)>::type;
    
    	// 将任务函数和其参数绑定，构建一个packaged_task
    	auto task = std::make_shared< std::packaged_task<return_type()> >(
    		std::bind(std::forward<F>(f), std::forward<Args>(args)...)
    		);
    	// 获取任务的future
    	std::future<return_type> res = task->get_future();
    	{
    		// 独占拿锁
    		std::unique_lock<std::mutex> lock(queue_mutex);
    
    		// don't allow enqueueing after stopping the pool
    		// 不允许入队到已经停止的线程池
    		if (stop) {
    			throw std::runtime_error("enqueue on stopped ThreadPool");
    		}
    		// 将任务添加到任务队列
    		tasks.emplace([task]() { (*task)(); });
    	}
    	// 发送通知，唤醒某一个工作线程取执行任务
    	condition.notify_one();
    	return res;
    }
    
    // the destructor joins all threads
    inline ThreadPool::~ThreadPool()
    {
    	{
    		// 拿锁
    		std::unique_lock<std::mutex> lock(queue_mutex);
    		// 停止标志置true
    		stop = true;
    	}
    	// 通知所有工作线程，唤醒后因为stop为true了，所以都会结束
    	condition.notify_all();
    	// 等待所有工作线程结束
    	for (std::thread &worker : workers) {
    		worker.join();
    	}
    }
    
    #endif