c++多线程2小时速成

简介

c++多线程基础需要掌握这三个标准库的使用：std::thread,std::mutex, andstd::async。

1. Hello, world

#include <iostream>
#include <thread>void hello() { std::cout << "Hello Concurrent World!\n"; 
}int main()
{std::thread t(hello);t.join();
}

• g++编译时须加上-pthread -std=c++11；
• 管理线程的函数和类在<thread>中声明，而保护共享数据的函数和类在其他头文件中声明；
• 初始线程始于main()，新线程始于hello()；
• join()使得初始线程必须等待新线程结束后，才能运行下面的语句或结束自己的线程；

2. std::thread

<thread> 和 <pthread> 都是用于多线程编程的库，但是它们针对不同的平台和编程环境。主要区别如下：

• 平台依赖性：

  • <thread> 是 C++11 标准库的一部分，提供了跨平台的多线程支持，因此可以在任何支持 C++11 标准的编译器上使用。
  • <pthread> 是 POSIX 线程库，它是在类 Unix 操作系统上的标准库，因此在 Windows 平台上并不直接支持。

• 语言级别：

  • <thread> 是 C++ 标准库的一部分，因此它提供了更加面向对象和现代化的接口，与 C++ 其他部分更加协同。
  • <pthread> 是 C 库的一部分，它提供了一组函数来操作线程，使用起来更加类似于传统的 C 风格。

• 使用方式：

  • <thread> 提供了 std::thread 类，使用起来更加符合 C++ 的面向对象设计，比如通过函数对象、函数指针等方式来创建线程。
  • <pthread> 提供了一组函数，比如 pthread_create() 来创建线程，它需要传入函数指针和参数。

下面来看看标准库thread类：

// 标准库thread类
class thread
{	// class for observing and managing threads
public:class id;typedef void *native_handle_type;thread() _NOEXCEPT{	// construct with no thread_Thr_set_null(_Thr);}template<class _Fn,class... _Args,class = typename enable_if<!is_same<typename decay<_Fn>::type, thread>::value>::type>explicit thread(_Fn&& _Fx, _Args&&... _Ax){	// construct with _Fx(_Ax...)_Launch(&_Thr,_STD make_unique<tuple<decay_t<_Fn>, decay_t<_Args>...> >(_STD forward<_Fn>(_Fx), _STD forward<_Args>(_Ax)...));}~thread() _NOEXCEPT{	// clean upif (joinable())_XSTD terminate();}thread(thread&& _Other) _NOEXCEPT: _Thr(_Other._Thr){	// move from _Other_Thr_set_null(_Other._Thr);}thread& operator=(thread&& _Other) _NOEXCEPT{	// move from _Otherreturn (_Move_thread(_Other));}thread(const thread&) = delete;thread& operator=(const thread&) = delete;void swap(thread& _Other) _NOEXCEPT{	// swap with _Other_STD swap(_Thr, _Other._Thr);}bool joinable() const _NOEXCEPT{	// return true if this thread can be joinedreturn (!_Thr_is_null(_Thr));}void join(){   // join threadif (!joinable())_Throw_Cpp_error(_INVALID_ARGUMENT);// Avoid Clang -Wparentheses-equalityconst bool _Is_null = _Thr_is_null(_Thr);if (_Is_null)_Throw_Cpp_error(_INVALID_ARGUMENT);if (get_id() == _STD this_thread::get_id())_Throw_Cpp_error(_RESOURCE_DEADLOCK_WOULD_OCCUR);if (_Thrd_join(_Thr, 0) != _Thrd_success)_Throw_Cpp_error(_NO_SUCH_PROCESS);_Thr_set_null(_Thr);}void detach(){	// detach threadif (!joinable())_Throw_Cpp_error(_INVALID_ARGUMENT);_Thrd_detachX(_Thr);_Thr_set_null(_Thr);}id get_id() const _NOEXCEPT{	// return id for this threadreturn (_Thr_val(_Thr));}static unsigned int hardware_concurrency() _NOEXCEPT{	// return number of hardware thread contextsreturn (_Thrd_hardware_concurrency());}native_handle_type native_handle(){	// return Win32 HANDLE as void *return (_Thr._Hnd);}

通过查看thread类的公有成员，我们得知：

• thread类包含三个构造函数：一个默认构造函数(什么都不做)、一个接受可调用对象及其参数的explicit构造函数(参数可能没有，这时相当于转换构造函数，所以需要定义为explicit)、和一个移动构造函数；
• 析构函数会在thread对象销毁时自动调用，如果销毁时thread对象还是joinable，那么程序会调用terminate()终止进程；
• thread类没有拷贝操作，只有移动操作，即thread对象是可移动不可拷贝的，这保证了在同一时间点，一个thread实例只能关联一个执行线程；
• swap函数用来交换两个thread对象管理的线程；
• joinable函数用来判断该thread对象是否是可加入的；
• join函数使得该thread对象管理的线程加入到原始线程，只能使用一次，并使joinable为false；
• detach函数使得该thread对象管理的线程与原始线程分离，独立运行，并使joinable为false；
• get_id返回线程标识；
• hardware_concurrency返回能同时并发在一个程序中的线程数量，当系统信息无法获取时，函数也会返回0。注意其是static修饰的，应该这么使用–std::thread::hardware_concurrency()。

由于thread类只有一个有用的构造函数，所以只能使用可调用对象来构造thread对象。

可调用对象包括：

• 函数
• 函数指针
• lambda表达式
• bind创建的对象
• 重载了函数调用符的类

3. join和detach

一旦您启动了您的线程，您需要明确地决定是等待它完成（通过join）还是让它自己运行（通过detach）。

使用 join() 的示例：

#include <iostream>
#include <thread>void threadFunction() {std::cout << "Thread is running..." << std::endl;// Simulate some workstd::this_thread::sleep_for(std::chrono::seconds(3));std::cout << "Thread completed." << std::endl;
}int main() {std::thread t(threadFunction);// 等待线程完成t.join();std::cout << "Main thread continues..." << std::endl;return 0;
}

在这个例子中，join() 方法用于等待线程 t 完成。主线程会被阻塞，直到线程 t 执行完毕。

运行结果：

使用 detach() 的示例：

#include <iostream>
#include <thread>void threadFunction() {std::cout << "Thread is running..." << std::endl;// Simulate some workstd::this_thread::sleep_for(std::chrono::seconds(3));std::cout << "Thread completed." << std::endl;
}int main() {std::thread t(threadFunction);// 分离线程，使得线程可以在后台运行t.detach();std::cout << "Main thread continues..." << std::endl;// 由于线程被分离，这里不等待线程完成，直接返回return 0;
}

在这个例子中，detach() 方法用于将线程 t 分离，使得线程可以在后台运行而不受主线程的控制。因此，主线程不会被阻塞，直接继续执行。

然后main已经结束

4. 线程间共享数据

条件竞争(race condition)：当一个线程A正在修改共享数据时，另一个线程B却在使用这个共享数据，这时B访问到的数据可能不是正确的数据，这种情况称为条件竞争。

数据竞争(data race)：一种特殊的条件竞争，并发的去修改一个独立对象。

多线程的一个关键优点(key benefit)是可以简单的直接共享数据，但如果有多个线程拥有修改共享数据的权限，那么就会很容易出现数据竞争(data race)。

std::mutex

C++保护共享数据最基本的技巧是使用互斥量(mutex)：访问共享数据前，使用互斥量将相关数据锁住，当访问结束后，再将数据解锁。当一个线程使用特定互斥量锁住共享数据时，其他线程要想访问锁住的数据，必须等到之前那个线程对数据进行解锁后，才能进行访问。

在C++中使用互斥量

• 创建互斥量：即构建一个std::mutex实例；
• 锁住互斥量：调用成员函数lock()；
• 释放互斥量：调用成员函数unlock()；
• 由于lock()与unlock()必须配对，就像new和delete一样，所以为了方便和异常处理，C++标准库也专门提供了一个模板类std::lock_guard，其在构造时lock互斥量,析构时unlock互斥量。

5. 异步好帮手：std::async

你可以使用 std::async 来在后台执行一个函数，并且可以通过 std::future 对象来获取函数的返回值。

#include <iostream>
#include <future>int taskFunction() {return 42;
}int main() {// 使用 std::async 异步执行任务std::future<int> future = std::async(std::launch::async, taskFunction);// 获取任务的结果int result = future.get();std::cout << "Result: " << result << std::endl;return 0;
}

另一个例子：

运行结果:

• std::async 接受一个带返回值的 lambda ，自身返回一个 std::future 对象。

• lambda 的函数体将在 另一个线程 里执行。

• 接下来你可以在 main 里面做一些别的事情， download 会持续在后台悄悄运行。

• 最后调用 future 的 get() 方法，如果此时 download 还没完成，会 等待 download 完成，并获取 download 的返回值。

6. 应用例子：线程池(thread pool)

如果你是一个程序员，那么你大部分时间会待在办公室里，但是有时候你必须外出解决一些问题，如果外出的地方很远，就会需要一辆车，公司是不可能为你专门配一辆车的，但是大多数公司都配备了一些公用的车辆。你外出的时候预订一辆，回来的时候归还一辆；如果某一天公用车辆用完了，那么你只能等待同事归还之后才能使用。

线程池(thread pool)与上面的公用车辆类似：在大多数情况下，为每个任务都开一个线程是不切实际的(因为线程数太多以致过载后，任务切换会大大降低系统处理的速度)，线程池可以使得一定数量的任务并发执行，没有执行的任务将被挂在一个队列里面，一旦某个任务执行完毕，就从队列里面取一个任务继续执行。

线程池通常由以下几个组件组成：

• 1.任务队列（Task Queue）：用于存储待执行的任务。当任务提交到线程池时，它们被放置在任务队列中等待执行。
• 2.线程池管理器（Thread Pool Manager）：负责创建、管理和调度线程池中的线程。它控制着线程的数量，可以动态地增加或减少线程的数量，以适应不同的工作负载。
• 3.工作线程（Worker Threads）：线程池中的实际执行单元。它们不断地从任务队列中获取任务并执行。
• 4.任务接口（Task Interface）：定义了要执行的任务的接口。通常，任务是以函数或可运行对象的形式表示。

使用线程池的好处

• 不采用线程池时：

  创建线程 -> 由该线程执行任务 -> 任务执行完毕后销毁线程。即使需要使用到大量线程，每个线程都要按照这个流程来创建、执行与销毁。

  虽然创建与销毁线程消耗的时间 远小于 线程执行的时间，但是对于需要频繁创建大量线程的任务，创建与销毁线程 所占用的时间与CPU资源也会有很大占比。

  为了减少创建与销毁线程所带来的时间消耗与资源消耗，因此采用线程池的策略：

  程序启动后，预先创建一定数量的线程放入空闲队列中，这些线程都是处于阻塞状态，基本不消耗CPU，只占用较小的内存空间。

  接收到任务后，线程池选择一个空闲线程来执行此任务。

  任务执行完毕后，不销毁线程，线程继续保持在池中等待下一次的任务。

  线程池所解决的问题：

  (1) 需要频繁创建与销毁大量线程的情况下，减少了创建与销毁线程带来的时间开销和CPU资源占用。（省时省力）

  (2) 实时性要求较高的情况下，由于大量线程预先就创建好了，接到任务就能马上从线程池中调用线程来处理任务，略过了创建线程这一步骤，提高了实时性。（实时）

简单的ThreadPool 类实现

参考GitHub - progschj/ThreadPool: A simple C++11 Thread Pool implementation

ThreadPool.h:

#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);template<class F, class... Args>auto enqueue(F&& f, Args&&... args) -> std::future<typename std::result_of<F(Args...)>::type>;~ThreadPool();
private:// need to keep track of threads so we can join themstd::vector< std::thread > workers;// the task queuestd::queue< std::function<void()> > tasks;// synchronizationstd::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([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(); });if(this->stop && this->tasks.empty())return;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;auto task = std::make_shared< std::packaged_task<return_type()> >(std::bind(std::forward<F>(f), std::forward<Args>(args)...));std::future<return_type> res = task->get_future();{std::unique_lock<std::mutex> lock(queue_mutex);// don't allow enqueueing after stopping the poolif(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);stop = true;}condition.notify_all();for(std::thread &worker: workers)worker.join();
}#endif

main.cpp:

#include <iostream>
#include <vector>
#include <chrono>#include "ThreadPool.h"int main()
{ThreadPool pool(4);std::vector< std::future<int> > results;for(int i = 0; i < 8; ++i) {results.emplace_back(pool.enqueue([i] {std::cout << "hello " << i << std::endl;std::this_thread::sleep_for(std::chrono::seconds(1));std::cout << "world " << i << std::endl;return i*i;}));}for(auto && result: results)std::cout << result.get() << ' ';std::cout << std::endl;return 0;
}

Basic usage:

// create thread pool with 4 worker threads
ThreadPool pool(4);// enqueue and store future
auto result = pool.enqueue([](int answer) { return answer; }, 42);// get result from future
std::cout << result.get() << std::endl;

这个 ThreadPool 类是一个简单的线程池实现，用于管理多线程任务的执行。下面逐步解释每个部分的功能：

1. 成员变量：

   • workers：一个 std::vector，存储所有工作线程。这些线程在构造函数中创建，并在析构函数中被回收。
   • tasks：一个 std::queue，用于存储等待执行的任务。
   • queue_mutex：一个互斥锁，用于保护任务队列，防止多个线程同时访问。
   • condition：条件变量，用于线程之间的协调，特别是在任务的等待和通知上。
   • stop：一个布尔值，用于告知线程何时停止执行，通常在析构函数中设置为 true。

2. 构造函数：

   • 构造函数初始化 stop 为 false，并创建指定数量的工作线程。每个线程执行一个无限循环，等待直到有任务可执行或接收到停止信号。线程通过锁定互斥锁和等待条件变量来同步，只有在有任务可执行或需要停止时才继续运行。
   • [this]: 这是一个 lambda 表达式，捕获当前 ThreadPool 对象的 this 指针，使得线程能够访问类成员变量如 queue_mutex、condition 和 tasks。
   • for(;;): 这是一个无限循环，用于不断地处理任务。线程将一直在这个循环中运行，直到接收到停止信号。
   • std::function<void()> task;: 定义一个 std::function 对象，用来存储从任务队列中取出的任务。
   • std::unique_lock<std::mutex> lock(this->queue_mutex);: 使用互斥锁保护代码块，确保在访问共享资源（任务队列）时不会有其他线程同时访问。
   • this->condition.wait(lock, [this]{ return this->stop || !this->tasks.empty(); });: 线程通过条件变量等待任务的到来或停止信号的到来。condition.wait 自动释放锁并进入等待状态，直到其他线程调用 notify_one() 或 notify_all() 时被唤醒，并在唤醒后重新获取锁。
   • if(this->stop && this->tasks.empty()) return;: 如果接收到停止信号，并且任务队列为空，则退出线程的执行函数，结束线程的运行。
   • task = std::move(this->tasks.front()); this->tasks.pop();: 将队列前端的任务移动到局部变量 task 中，并从队列中移除该任务。
   • task();: 执行从队列中取出的任务。

3. enqueue() 方法：

   • 这个 enqueue 方法提供了一种强大的机制来将任何可调用的任务安排到线程池中，并通过 std::future 提供了一种获取异步执行结果的方法。
   • template<class F, class... Args>: 这是一个模板声明，表示这个函数可以接受任何类型的函数 F 和任意数量和类型的参数 Args...。
   • auto ThreadPool::enqueue(F&& f, Args&&... args): enqueue 函数接受一个可调用对象 f 和它的参数 args...，这些参数都是通过完美转发（std::forward）传递的，以保持其值类别（lvalue 或 rvalue）。
   • -> std::future<typename std::result_of<F(Args...)>::type>: 函数的返回类型是 std::future，其中包含了函数 f 执行后的返回类型。std::result_of<F(Args...)>::type 在 C++17 之前用于推断函数的返回类型。
   • using return_type = typename std::result_of<F(Args...)>::type;: 这一行定义了 return_type，即函数 f 的返回类型。
   • std::make_shared< std::packaged_task<return_type()> >(...): 创建一个 std::packaged_task 对象，该对象封装了函数 f 及其参数，并用 std::shared_ptr 管理。std::packaged_task 被设计用来执行并保存其返回值（或异常），这个返回值可以通过与之关联的 std::future 对象获得。
   • std::bind(std::forward<F>(f), std::forward<Args>(args)...): 使用 std::bind 将函数 f 和它的参数绑定起来，通过 std::forward 完美转发参数，确保参数的正确传递（包括保持其lvalue或rvalue性质）。
   • std::future<return_type> res = task->get_future();: 从 std::packaged_task 获取一个 std::future 对象，该对象将在任务完成时保存结果。
   • std::unique_lock<std::mutex> lock(queue_mutex);: 在修改任务队列之前，先锁定互斥量以确保线程安全。
   • if(stop) throw std::runtime_error("enqueue on stopped ThreadPool");: 如果线程池已经停止，则不允许再添加新任务，抛出异常。
   • tasks.emplace([task](){ (*task)(); });: 将任务作为一个 lambda 表达式添加到任务队列。这个 lambda 包裹了 std::packaged_task 的调用。
   • condition.notify_one();: 通知一个正在等待的线程（如果有的话），告诉它现在有新的任务可以执行。
   • return res;: 返回与任务关联的 std::future 对象，允许调用者获取异步执行的结果。

4. 析构函数：

   • 析构函数会首先设置 stop 标志位为 true，然后通过条件变量 condition.notify_all() 通知所有等待的工作线程。
   • 最后，使用 join() 方法等待所有工作线程执行完毕

应用场景：

1. Web服务器： Web服务器通常需要处理大量并发的客户端请求。每个请求可能包括计算任务，如处理HTTP请求、访问数据库或生成动态内容。线程池可以用来管理这些并发任务，提高响应速度和资源利用率。

2. 数据处理： 在数据密集型应用程序中，如数据分析或图像处理，任务通常可以并行处理。使用线程池可以将数据分块处理，每个线程处理一块数据，从而加速整体处理速度。

3. 异步I/O操作： 在需要执行大量异步I/O操作的应用中，例如文件处理或网络通信，线程池可以用来管理这些异步操作，提高处理效率。

4. GUI应用程序： 图形用户界面（GUI）应用程序需要快速响应用户交互，同时执行后台任务如加载数据或执行计算。线程池可以在后台处理这些任务，保持界面的响应性和流畅性。

示例代码

假设我们正在开发一个简单的图片处理应用，需要对一组图片进行大小调整。使用线程池可以并行处理每张图片，提高处理速度。

#include <iostream>
#include <vector>
#include <chrono>
#include "ThreadPool.h"// 假设的图片处理函数，这里只是简单模拟处理时间
void processImage(int imgSize) {std::this_thread::sleep_for(std::chrono::milliseconds(100));  // 模拟耗时操作std::cout << "Processed image of size " << imgSize << " in thread " << std::this_thread::get_id() << std::endl;
}int main() {ThreadPool pool(4);  // 创建一个有4个工作线程的线程池std::vector<std::future<void>> results;std::vector<int> imageSizes = {1024, 2048, 512, 128, 256, 768};  // 各种大小的图片for (int size : imageSizes) {results.emplace_back(pool.enqueue(processImage, size));}// 等待所有图片处理完成for (auto &&result : results) {result.get();  // 获取任务结果（这里实际没有返回值）}std::cout << "All images processed." << std::endl;return 0;
}

在这个示例中，我们创建了一个线程池并并行处理了多张图片。每张图片的处理是通过调用 processImage 函数完成的，该函数接受一个代表图片大小的整数参数。我们通过线程池的 enqueue 方法提交了多个任务，并使用 std::future<void> 来跟踪这些任务的完成状态。处理完成后，主函数等待所有任务完成，然后输出完成信息。

参考：

https://chorior.github.io/2017/04/24/C++-thread-basis/#managing_threads

C++ Concurrency In Action，Anthony Williams

GitHub - progschj/ThreadPool: A simple C++11 Thread Pool implementationx

线程池代码解读：C++线程池实现解析_template<class f,class...args>-CSDN博客