C++多线程

Posted 2021-04-21 CPP编程客

关于多线程的这几篇是连着写的，理清了Win32中的多线程再来看C++的多线程会轻松许多，联系着看能更好的理解其中明细所以。

现在来看看C++的多线程吧。

---

Thread

直到C++11才提供了一些线程相关操作，其中有一个std::thread类用于启动及处理线程。其用法简单，使用起来就像这样：

1void ThreadFunc(int param);
2
3int num = 20;
4std::thread t(ThreadFunc, num);
5
6t.join();

再来看看Win32的线程，写出对应的如下：

1void ThreadFunc(LPVOID param);
2
3int num = 20;
4HANDLE hThread = CreateThread(NULL, 0, 
5                        (LPTHREAD_START_ROUTINE)ThreadFunc, (void *)num, 0, NULL);
6
7WaitForSingleObject(hThread, INFINITE);

使用起来是不是方便使用多了？

std::thread类提供了线程的基本操作，成员函数join()和传入INFINITE的WaitForSingleObject一样，用于等待当前关联的线程执行完毕。

当t有一个关联函数的时候，表示处于joinable（可连接）状态，可调用成员函数 joinable()查看。当处于joinable状态时便可用join()连接等待关联函数完成工作，完成后将变为nonjoinable状态。若线程不是joinable状态，用join()连接将会抛出std::system_error异常。

这个和Win32中的signaled状态相似，当内核对象是signaled状态的时候WaitForSingled才能等待线程执行，线程执行完了则会变为nonsignaled状态。同样，nonjoinable状态也表示线程已执行完毕了，处于不可连接（等待）状态。

除了join()处，还有一个detach()成员函数用于分离线程，当调用的时候会解除t与线程之间的关联，但是线程依旧会继续运行。若线程不是joinable状态，依旧会抛出std::system_error异常。

基本上我们都应使用join()而非detach()。因为分离的线程我们将会对其失去控制，从而无法得知它是否运行，也无法得知它运行了多久，当线程访问一些资源的时候，可能资源早已失效，从而导致不可预料的行为。

既然是个线程类，那肯定还提供的有一些线程信息的获取函数。std::thread提供了一个get_id()用于获取线程ID，在CreateThread函数中我们是通过最后一个参数获取的。

这个get_id()会在joinable状态时返回一个独一无二的std::thread::id，nonjoinable时会返回一个std::thread::id()，表示非线程ID，非线程ID是由std::thread的default构造函数产出的一个特殊值。

std::thread还提供有一个static成员函数std::thread::hardware_concurrency()用于获取当前CPU支持的线程数，这个和使用GetSystemInfo获取中的dwNumberOfProcessors是一样的:

1SYSTEM_INFO si;
2GetSystemInfo(&si);
3std::cout << si.dwNumberOfProcessors << std::endl;

std::thread位于< thread>中，在这个头文件中还提供了一个std::this_thread，它针对任何线程提供了线程专属的全局函数。

this_thread::get_id()用于获取当前线程的id，其中还提供两个用于阻塞的函数：

• this_thread::sleep_for()

• this_thread::sleep_until()

在win32中我们是通过Sleep()来阻塞的，Sleep()和sleep_for()功能一样，只是sleep_for()使用的参数为std::chrono时间库，时间更加精准。

而sleep_until()用于指定阻塞结束的时间点，对于阻塞操作更加灵活。

其中还有一个this_thread::yield()，用于释放对于当前线程的调度，告诉系统放弃当前线程的时间片，使其它线程得以执行。

来看一个小例子：

 1#include <thread>
 2#include <iostream>
 3#include <Windows.h>
 4#include <chrono>
 5
 6
 7void ThreadFunc(int param)
 8{
 9    for (int i = 0; i < param; ++i)
10    {
11        //等待100ms
12        std::this_thread::sleep_for(std::chrono::milliseconds(100));
13        std::cout.put('0' + i).flush();
14    }
15    std::cout << std::endl;
16}
17
18
19int main()
20{
21    int num = 20;
22    std::thread t(ThreadFunc, num);
23
24    std::cout << std::boolalpha << "joinable: " << t.joinable() << std::endl;
25    std::cout << "CPU支持线程数: " << std::thread::hardware_concurrency() << std::endl;
26
27    t.join();
28    std::cout << std::boolalpha << "joinable: " << t.joinable() << std::endl;
29
30    std::cin.get();
31    return 0;
32}

输出如下：

1joinable: true
2CPU支持线程数: 8
30123456789:;<=>?@ABC
4joinable: false

---

Async And Future

C++还提供了一个高级接口std::async()，也可用来开启线程。

std::async对于线程提供了提供了更多的控制，它可以处理线程返回的结果，可以处理异常，指定发射策略，还是以一个例子来看：

 1#include <future>
 2#include <iostream>
 3
 4int ThreadFunc(char ch, int times)
 5{
 6    for (int i = 0; i < times; ++i)
 7    {
 8        std::cout.put(ch).flush();
 9    }
10
11    std::cout << std::endl;
12
13    return ch;
14}
15
16int main()
17{
18    std::future<int> fu(std::async(ThreadFunc, '+', 3));
19    int result = ThreadFunc('-', 5);
20
21    std::cout << "return values: " << result << " " 
22                            << fu.get() << std::endl;
23
24    std::cin.get();
25    return 0;
26}

这里使用std::async异步开启了一个线程调用ThreadFunc，它会产生一个结果std::future，称为期货，因为开始时并不知道明确值，这是个未来结果。

1std::future<int> fu(std::async(ThreadFunc, '+', 3));

这里std::future 也可以用auto来推导出结果，所以也可以这么写：

1auto fu(std::async(ThreadFunc, '+', 3));

接着是一个普通的函数调用，返回值存到result中：

1int result = ThreadFunc('-', 5);

然后调用输出，可以通过std::funture的get函数获取目标函数返回的值。std::async调用便会立即尝试启动ThreadFunc，所以调用get()时也许ThreadFunc已经启动于一个分离线程并且已执行完毕，这时会直接输出结果；如果调用get时ThreadFunc还未执行完毕，get便会一直阻塞直到其执行完毕；若是调用get时ThreadFunc尚未启动，那么会强迫调用并阻塞以获得结果。

1std::cout << "return values: " << result << " " 
2                            << fu.get() << std::endl;

这样就保证了在调用get()后无论目标函数是异步还是同步执行，线程都已执行完毕。

所以程序的输出可能如下：

1-++-+---
2
3return values: 45 43

若是线程函数不返回任何数，那么得到的期货是一个偏特化版本的std::future< void>，这种情况下调用get就只是等待线程函数执行完毕。

当不调用get()时，那么ThreadFunc是不一定会被调用的，这时还可以使用发射策略来保证线程的执行。

1auto fu = std::async(std::launch::async, ThreadFunc, '+', 3);

当传入std::launch::async后，std::async就不会推延线程函数的执行，这就会像同步调用一样。若是无法调用，那么会抛出一个std::system_error异常，当调用get()时，该异常会被抛出来，我们可以处理它：

1try {
2    fu.get();
3}
4catch(const std::exception& e)
5{
6    std::cerr << "Exception: " << e.what() << std::endl;
7}

也可以强制它延缓执行，只需传入std::launch::deferred,那么就直到调用get时才会执行目标函数。

而默认使用的未指定策略的形式：

1auto fu(std::async(ThreadFunc, '+', 3));

它会视情况来决定使用std::launch::async还是std::launch_deferred。

注意的是std::future只能get()一次，之后std::future就处于无效状态了，可以用valid()来检测它的有效性：

1auto fu = std::async(std::launch::async, ThreadFunc, '+', 3);
2
3assert(fu.valid());
4std::cout << fu.get() << std::endl;
5assert(!fu.valid());

那么我们要想多次处理返回结果呢？std::async提供了一个std::shared_future，future和shared_future的关系就像unique_ptr和shared_ptr，自然的就联想到摧毁式拷贝和引用计数，shared_future通常的实现就是一个引用计数，当最后一个使用者使用后才被销毁，关于这个可以看内存管理之智能指针。

关于shared_future的使用可以看这个小例子：

 1#include <future>
 2#include <iostream>
 3#include <cassert>
 4#include <string>
 5
 6void ThreadFunc(std::string str, const std::shared_future<std::string>& sf)
 7{
 8    std::cout << str << sf.get() << std::flush << std::endl;
 9}
10
11std::string SharedString()
12{
13    std::string str;
14    std::cin >> str;
15    return str;
16}
17
18int main()
19{
20    std::shared_future<std::string>  sf = std::async(std::launch::async, SharedString);
21
22    std::async(std::launch::async, ThreadFunc, "sf1-", sf);
23    std::async(std::launch::async, ThreadFunc, "sf2-", sf);
24    std::async(std::launch::async, ThreadFunc, "sf3-", sf);
25
26    std::cin.get();
27    return 0;
28}

关于std::shared_future，若也想用auto推断类型的话得在后面加个share()：

1auto sf = std::async(std::launch::async, SharedString).share();

测试输出为：

1running
2sf1-running
3sf2-running
4sf3-running

使用std::future或std::shared_future时，要是我们不想知道返回结果，只想等待目标函数执行终止，可以调用wait()成员函数。

相应的，它也提供了基于时间段和时间点的等待函数wait_for()和wait_until()，和std::this_thread::sleep_for()与std::this_thread::sleep_until()一样，也使用的std::chrono时间库。

wait_for()和wait_until()有三种返回情况:

• std::future_status::deferred

• std::future_status::timeout

• std::future_statur::ready

指定了std::launch::deferred而又未调用wait()或get()以强制启动的情况下会返回第一种情况。

在指定的时间段或时间点结束后，异步执行的操作还未结束，就会返回timeout。

所有操作完成后，便会返回第三种情况ready。

可以这样获取它的状态：

1std::future_status s = sf.wait_for(std::chrono::milliseconds(1000));

---

Promise

std::promise用于在线程中传递一个返回值或异常，用法比较灵活。std::future便是基于这个实现的，其中主要有两个方法：

• set_value  用于传递数值

• set_exception  用于传递异常

可以使用这个来在std::thread中传递数值或异常。

 1#include <iostream>
 2#include <exception>
 3#include <future>
 4#include <thread>
 5#include <string>
 6
 7void ThreadFunc(std::promise<std::string>& p)
 8{
 9    try {
10        std::string ep;
11        std::getline(std::cin, ep);
12        //若输入exception则抛出异常
13        if (ep.compare("exception") == 0)
14        {
15            throw std::runtime_error("Thread func exception");
16        }
17
18        //若是其它值，则设为返回值
19        p.set_value(ep);
20    }
21    catch (...)
22    {
23        //设置异常
24        p.set_exception(std::current_exception());
25    }
26}
27
28int main()
29{
30
31    try {
32        std::promise<std::string> p;
33        std::thread t(ThreadFunc, std::ref(p));
34        t.detach();
35
36        //获取设置的值
37        std::future<std::string> f(p.get_future());
38
39        std::cout << "result: " << f.get() << std::endl;
40    }
41    catch (const std::exception& e)
42    {
43        std::cerr << "EXCEPTION: " << e.what() << std::endl;
44    }
45    catch (...)
46    {
47        std::cerr << "EXCEPTION: " << std::endl;
48    }
49
50    std::cin.get();
51    return 0;
52}

当正常返回时：

当触发异常时：

若是想在线程退出时设置数据或异常，可以调用std::promise的：

• set_value_at_thread_exit()

• set_exception_at_thread_exit()

需要注意的是同时只能设置一个，设置多个的话会导致异常。

C++线程的基本使用就是这些，我们可以使用thread配合promise使用，也可以使用async配合future来使用。

关于同步，其实也和win32中的大同小异，下篇来说。