文章

c++(2·多线程)

发表于 更新于
作者 senlongan
10 分钟阅读

多线程

基础概念

  • 进程：一个正在运行的应用程序被称为一个进程
  • 线程：是进程中的实际运行单位，每个进程可以拥有至少一个的线程
  • 同步：一个任务完成后另一个任务才能开始
  • 异步：一个任务的开始和完成不会直接影响另一个任务的开始和完成
  • 并发：有多个任务在单核CPU被处理，任务被交替执行
  • 并行：有多个任务在多核CPU被处理，任务可以被同时执行

多线程

多线程是并发/并行的技术实现，使得程序能够在同一时间执行多于一个任务，进而提升整体处理性能，但有时它并非一定会提升性能

  • 单核处理CPU任务：多线程会时间切片，交替执行任务，这是同步且并发的，CPU任务不能被异步处理，且任务调度需要时间，因此不会提高效率
  • 单核处理CPU + GPU任务：由于GPU是高度并行的，因此可以在执行CPU任务时同时处理多个GPU任务，CPU和GPU两者之间是异步且并行的，会提高效率
  • 多核处理CPU任务：多个任务被异步处理，这是异步且并行的，会提高效率

std::thread

1
2
3
4
5
6
7
8
9
10
11
12
13
14

#include <thread>

thread a(
    []{
        cout << "Hello, " << flush;
    }
)//创建线程a并执行函数
a.join();//线程结束后清理

void printId(int id) {
    std::cout << id << std::endl;
}
thread th(print, 1);//有参线程
th.join();

展开

  • thread()创建一个线程
  • thread(Fn&& fn, Args&&… args)创建线程并执行函数（可调用对象），Args是thread模板类的模板参数列表，&&右值引用，组合为万能引用，… 可变数量的参数（>=0个）
  • thread(thread&& x) 移动构造（注意没有拷贝构造，否则会有严重的混乱问题）
  • join()阻塞当前线程（在哪个线程调用） 后续的指令，直到目标线程执行完毕，线程结束清理资源（像指针的delete），它能确保线程执行的完整性
  • detach（）将线程与当前线程分离，彼此独立执行，也就是不会阻塞当前线程，但要注意如果主程序结束了，这些分离线程会被强制终止，不能确保线程执行的完整性，会自动清理资源
  • 注：如果没有调用join 或 detach，程序的行为是未定义的，可能会内存泄漏等问题
  • joinable()返回线程是否可以执行join函数，这避免了对已经 join 或 detach 过的线程再次操作，否则会崩溃
  • std::this_thread::get_id()获取主线程id，t1.get_id()获取新线程id
  • std::this_thread::sleep_for(std::chrono::seconds(2))主线程休眠2秒
  • std::\thread::hardware_concurrency()获取CPU核心数/线程数
  • yield()让步 暂时放弃线程的执行，将主动权 / 时间片 交给其他线程
  • 应用：在图形学像素处理时，将像素按照区域均分给每个线程处理，由于没有共享变量，因此数据安全

引用参数

1
2
3
4
5
6
7

void changevalue(int &b, int val) {
	b = val;
}
int a = 0;
thread th(changevalue, a, 5);//❌：
thread th(changevalue, ref(a), 5);//✅
th.join();

展开

但是thread为了确保线程的执行独立性，任何情况外部参数都会强制拷贝到线程内部存储，无论是否被推导为左值引用，并将右值临时对象传递给函数，由于即是拷贝又是右值，因此传递给左值引用将发生错误

std::ref（x）和std::cref（x）使得a传入thread时不会发生拷贝，因此左值引用接受左值，是没问题的

std::atomic和std::mutex

数据竞争：指多个线程同时操作同一个共享变量，并且至少有一个访问是写操作，其他为读操作

为了应对这种情况，使用：

1
2
3
4
5
6
7
8
9
10
11

std::mutex counter_mutex;//创建互斥锁
counter_mutex.lock(); // 锁定互斥锁
x++; // 共享变量
counter_mutex.unlock(); // 解锁

std::mutex counter_mutex;//创建互斥锁
std::lock_guard<std::mutex> lock(counter_mutex);
x++; // 共享变量

std::atomic<int> x = 0; //或者别名 atomic_int x = 0;
x++; // 共享变量

展开

  • mutex互斥琐：
    • 通过mutex将共享变量定义为互斥量，当一个线程将mutex lock()锁住时，其它的线程就不能操作mutex，其他线程会被阻塞，直到这个线程将mutex unlock()解锁，
    • 可以用try_lock()查看mutex是否被锁住，
    • 如果忘记unlock或者发生异常，就会死锁，更推荐使用RAII风格的std::lock_guard，
    • mutex在每次被修改都要上锁、解锁，因此效率很低
  • atomic原子：通过atomic< 数据类型 > 定义一个原子，表明它不能异步执行，需要被同步操作，因此免去了mutex每次上锁、解锁的时间消耗

std::async

  • async (Fn&& fn, Args&&… args)开始执行fn函数，以args为参数，返回std::future
  • async (launch policy, Fn&& fn, Args&&… args);其中launch是枚举，表示启动策略
  • launch::async异步启动
  • launch::deferred同步启动

async vs. thread

  • async是函数而非类，返回结果为std::future未来
  • async可以选择启动策略
  • async会自动管理线程的生命周期，也就是不需要join() / detach()释放资源
  • thread适用于对线程进行精细控制的场景，async适用于简单的线程任务

std::future

1
2
3
4
5
6

template<class ... Args> 
decltype(auto) sum(Args&&... args) {
	return (0 + ... + args);
}
future<int> val = async(launch::async, sum<int, int, int>, 1, 10, 100);
cout << val.get() << endl;

展开

  • 创建一个future<对应函数返回类型>类型的变量等于async函数的结果，还可以用来检测线程是否已结束
  • get()等待线程结束并获取返回值
  • wait() 阻塞当前线程并等待新线程结束
  • 检查线程是否结束wait_for（rel_time）阻塞等待rel_time一段时间，如果在这段时间内新线程结束则返回future_status::ready已完成状态（强枚举类），若没结束则返回future_status::timeout返回超时状态，若async是以launch::deferred启动的，则不会阻塞并立即返回future_status::deferred

std::promise

thread通过引用获取返回值

1
2
3
4
5
6
7
8
9
10
11

void calculate_sum(const std::vector<int>& nums, int& result) {
    result = 0;
    for (int num : nums) result += num;
}
int main() {
    std::vector<int> nums = {1, 2, 3, 4, 5};
    int sum = 0;
    std::thread t(calculate_sum, std::ref(nums), std::ref(sum));
    t.join();
    return 0;
}

展开

由于thread类并没有提供直接获取返回值的函数，想要获取thread线程的返回值，可以通过引用获取返回值

thread通过promise获取返回值

1
2
3
4
5
6
7
8
9
10
11
12
13
14

void calculate_sum(const std::vector<int>& nums, std::promise<int> result_promise) {
    int sum = 0;
    for (int num : nums) sum += num;
    result_promise.set_value(sum); // 设置结果
}

int main() {
    std::vector<int> nums = {1, 2, 3, 4, 5};
    std::promise<int> sum_promise;
    std::thread t(calculate_sum, std::ref(nums), ref(sum_promise));
	cout << sum_promise.get_future().get() << endl;
    t.join(); // 可异步执行
    return 0;
}

展开

promise承诺 实际上是std::future的一个包装，

  • promise()
  • promise(allocator_arg_t aa, const Alloc& alloc)使用特定的内存分配器alloc构造对象
  • promise (promise&& x) 移动构造
  • get_future()构造一个future对象，其值与promise相同
  • set_value （const T& val / T&& val）设置值
C++
本文由作者按照 CC BY 4.0 进行授权
本页总访问量次