Logo

菜单

蔚淦丞
发布于 2026-08-31 / 12 阅读
0
0

C++多线程

C++多线程

state: Doing

锁

lambda函数的写法

#include<iostream>
#include<thread>
#include<chrono>
#include<mutex>

using namespace std;

int main() 
{
	// 1.atomic<int> count 非通用解决方案，如果是基本类型可以这样做，解决线程处理同一个数据的问题
	int count = 0;
	const int ITERATIONS = 100000;

	// 2.mutex加锁解锁，lock(), unlock()用在临界区前后
	mutex mtx;

	auto func = [&]() {
		for (int i = 0; i < ITERATIONS; i++)
		{
			mtx.lock();
			++count;
			mtx.unlock();
		}
	};

	thread t1(func); // 创建线程 直接thread 线程名，创建以后应该就开始执行了
	thread t2(func);
	t1.join();       // 必须调用join()，join会挂起调用线程(main函数的线程)，等待被调用线程(t1线程)执行结束，才会调用主线程继续执行。
	t2.join();

	cout << count << endl;

	return 0;
}

带参数的函数的写法

#include<iostream>
#include<thread>
#include<chrono>
#include<mutex>

using namespace std;

void work(int &count, mutex &mtx) {
	for (int i = 0; i < 1e6; i++)
	{
		mtx.lock();
		++count;
		mtx.unlock();
	}
};

int main() 
{
	int count = 0;
	mutex mtx;

	**thread t1(work, ref(count), ref(mtx)); //第一个参数是函数名，函数后面带上参数就行，传递引用要写ref()
	thread t2(work, ref(count), ref(mtx));**
	t1.join();
	t2.join();

	cout << count << endl;
	return 0;
}

lock_guard 和 unique_lock

如果直接用mtx.lock() mtx.unlock()，当临界区有exception抛出，就不会unlock，所以要用lock_guard 和 unique_lock

lock_guard:

  1. 没有提供加锁和解锁的接口
  2. 通过构造函数和析构函数控制锁的作用范围，创造对象的时候加锁，离开作用域的时候解锁
  3. 稍微笨重一点，但对资源消耗要小一点

unique_lock：

  1. 提供了lock和unlock接口，能记录现在处于上锁还是没上锁状态
  2. 可以通过构造函数和析构函数控制锁的范围
    在析构函数中延时加锁，在需要的时候手动枷锁和解锁
    在析构的时候，会根据当前状态来决定是否要进行解锁（lock_guard就一定会解锁）
  3. unique_lock更加灵活，因为他维持mutex的状态，但也因此对于资源的消耗更大，同时效率也更低
#include<iostream>
#include<thread>
#include<chrono>
#include<mutex>

using namespace std;

void work(int &count, mutex &mtx) {
	for (int i = 0; i < 1e6; i++)
	{
		**// 1. lock_guard**
		**// lock_guard<mutex> guard(mtx); // 用lock_guard<mutex> guard(mtx) 代替前面的 mtx.lock() ... mtx.unlock() 在其作用域是临界区

		// 2. unique_lock 两者非常相似
		unique_lock<mutex> uniLock(mtx);**
		++count;
	}
};

int main() 
{
	int count = 0;
	mutex mtx;

	thread t1(work, ref(count), ref(mtx));
	thread t2(work, ref(count), ref(mtx));
	t1.join();
	t2.join();

	cout << count << endl;

	return 0;
}

函数对象（仿函数）的写法

对于类，可以把count和mtx写为成员数据

#include<iostream>
#include<thread>
#include<chrono>
#include<mutex>

using namespace std;

class App
{
private:
	int count = 0;
	mutex mtx;
public:
	void operator()() {
		for (int i = 0; i < 1e6; i++)
		{
			// lock_guard<mutex> guard(mtx); // 用lock_guard<mutex> guard(mtx) 代替前面的 mtx.lock() ... mtx.unlock()
			unique_lock<mutex> uniLock(mtx);
			++count;
		}
	}
	int getCount() {
		return count;
	}
};

int main() 
{
	App app;
	**thread t1(ref(app)); // 这里必须传递app对象的引用进去，因为mutex不允许拷贝**
	**thread t2(ref(app));**
	t1.join();
	t2.join();

	cout << app.getCount() << endl;
	return 0;
}

从线程返回数据

future

如果用promise，需要set_value,就得重写子线程要调用的函数，用auto…类似于闭包的方法把原函数包起来，然后set_value（return的值）

promise传给其他线程，用来set_value()

future接收其他线程返回的值future = promise.get_future()

#include<iostream>
#include<thread>
#include<cmath>
#include<iomanip>
#include<future>

using namespace std;

double calculate_pi(int terms) {
	double res = 0.0;
	for (int i = 0; i < terms; i++) {
		int sign = pow(-1, i);
		double term = 1.0 / (i * 2 + 1);
		res += sign * term;
	}
	return res * 4;
}

int main() 
{
	**promise<double> promise;**		// future库的promise<...> 可以在main线程接收子线程的值

	auto do_calculation = [&](int terms) { // 所有的值都要用到，所以可以其实是把promise也传进去了
		auto result = calculate_pi(terms);
		**promise.set_value(result);**
	};

	thread t1(do_calculation, 1e6);

	**future<double> future = promise.get_future();**
	cout << setprecision(15) << **future.get()** << endl; // **future.get()会阻塞主进程**，直到t1线程promise.set_value(),所以t1.join()写在下面也无所谓

	t1.join();
	return 0;
}

exception

有exception的情况同样用future.get来获取，只是promise会调用set_exception 而不是 set_value，同样future获取到的是exception而不是value

promise set 和 future get 都用try catch 模块来获取。

#include<iostream>
#include<thread>
#include<cmath>
#include<iomanip>
#include<future>
#include<exception>

using namespace std;

double calculate_pi(int terms) {
	double res = 0.0;

	if (terms < 1) {
		throw runtime_error("Terms cannot be less than 1");
	}

	for (int i = 0; i < terms; i++) {
		int sign = pow(-1, i);
		double term = 1.0 / (i * 2 + 1);
		res += sign * term;
	}
	return res * 4;
}

int main() 
{
	promise<double> promise;		// future库的promise<...> 可以在main线程接收子线程的值

	auto do_calculation = [&](int terms) { // 所有的值都要用到，所以可以看出把promise也传进去了
		**try
		{
			auto result = calculate_pi(terms);
			promise.set_value(result);
		}
		catch (...)
		{
			promise.set_exception(current_exception());
		}**
	};

	thread t1(do_calculation, 1e6);
	future<double> future = promise.get_future();

	**try
	{
		cout << setprecision(15) << future.get() << endl;
	}
	catch (const std::exception& e)
	{
		cout << e.what() << endl;
	}**

	t1.join();

	return 0;
}

packaged tasks

用packagee_task 可以简单的接收线程返回值，不用像那样设置promise，也不用类似闭包的办法把函数包起来，再传值

#include<iostream>
#include<thread>
#include<cmath>
#include<iomanip>
#include<future>
#include<exception>

using namespace std;

double calculate_pi(int terms) {
	double res = 0.0;

	if (terms < 1) {
		throw runtime_error("Terms cannot be less than 1");
	}

	for (int i = 0; i < terms; i++) {
		int sign = pow(-1, i);
		double term = 1.0 / (i * 2 + 1);
		res += sign * term;
	}
	return res * 4;
}

int main() 
{
	**packaged_task<double(int)> task1(calculate_pi);**

	**future<double> future1 = task1.get_future();**

	thread t1(move(task1), 1e6);

	**double result = future1.get();**

	cout << setprecision(15) << result << endl;

	t1.join();

	return 0;
}

exception也可以接收

#include<iostream>
#include<thread>
#include<cmath>
#include<iomanip>
#include<future>
#include<exception>

using namespace std;

double calculate_pi(int terms) {
	double res = 0.0;

	if (terms < 1) {
		throw runtime_error("Terms cannot be less than 1");
	}

	for (int i = 0; i < terms; i++) {
		int sign = pow(-1, i);
		double term = 1.0 / (i * 2 + 1);
		res += sign * term;
	}
	return res * 4;
}

int main() 
{
	packaged_task<double(int)> task1(calculate_pi);

	future<double> future1 = task1.get_future();

	thread t1(move(task1), 0);

	try
	{
		double result = future1.get();
		cout << setprecision(15) << result << endl;
	}
	catch (const std::exception& e)
	{
		cout << "ERROR! " << e.what() << endl;
	}

	t1.join();

	return 0;
}

线程同步

condition_variable

  1. 线程1构造condition variable
  2. 线程2运行完，执行variable.notify_one/all
  3. 线程1中variable.wait(lock)，在线程2执行notify唤醒后，返回。
#include<iostream>
#include<thread>
#include<chrono>
**#include<condition_variable>**

using namespace std;

int main() 
{
	**condition_variable condition; //构建condition_variable**
	mutex mtx;
	bool ready = false;

	thread t1([&]() {
		this_thread::sleep_for(chrono::milliseconds(2000)); //t1线程sleep 2秒
		**unique_lock<mutex> lock(mtx); //必须用unique_lock，因为需要unlock方法，而lock_guard没有提供unlock方法**
		ready = true;
		**lock.unlock();
		condition.notify_one();  //lock.unlock()后 condition variable调用notify方法，唤醒主线程**
	});

	unique_lock<mutex> lock(mtx);
	while (!ready) {
		**condition.wait(lock);  //主线程在这里进入等待，等待t1线程唤醒**
	}

	cout << "ready " << ready << endl;

	t1.join();

	return 0;
}

小改进 不用while 等待

#include<iostream>
#include<thread>
#include<chrono>
#include<condition_variable>

using namespace std;

int main() 
{
	condition_variable condition;
	mutex mtx;
	bool ready = false;

	thread t1([&]() {
		this_thread::sleep_for(chrono::milliseconds(2000)); //t1线程sleep 2秒

		cout << "t1 acquiring lock" << endl;
		unique_lock<mutex> lock(mtx);
		cout << "t1 acquired lock" << endl;
		ready = true;
		lock.unlock();
		cout << "t1 release lock; notifying" << endl;
		condition.notify_one();
	});

	cout << "main acquiring lock" << endl;
	unique_lock<mutex> lock(mtx);

	cout << "main acquired lock; waiting" << endl;
	condition.wait(lock, [&]() { return ready; });

	cout << "main finished waiting" << endl;
	cout << "ready " << ready << endl;

	t1.join();

	return 0;
}

阻塞队列

#include<iostream>
#include<thread>
#include<queue>
#include<mutex>
#include<condition_variable>

using namespace std;

template<typename E>
class blocking_queue
{
private:
	mutex _mtx;
	condition_variable _cond;
	int _max_size;
	queue<E> _queue;

public:
	blocking_queue(int max_size): _max_size(max_size)
	{

	}

	void push(E e) {
		unique_lock<mutex> lock(_mtx);
		_cond.wait(lock, [this]() { return _queue.size() < _max_size; });
		_queue.push(e);
		lock.unlock();
		_cond.notify_one();
	}

	void pop() {
		unique_lock<mutex> lock(_mtx);
		_cond.wait(lock, [this]() { return !_queue.empty(); });

		_queue.pop();

		lock.unlock();
		_cond.notify_one();
	}

	E front() {
		unique_lock<mutex> lock(_mtx);
		_cond.wait(lock, [this]() { return !_queue.empty(); });
		return _queue.front();
	}

	int size() {
		lock_guard<mutex> lock(_mtx);
		return _queue.size();
	}
};

int main()
{
	blocking_queue<int> qu(2);

	thread t1([&]() {
		for (int i = 0; i < 10; i++)
		{
			cout << "pushing " << i << endl;
			cout << "queue size is " << qu.size() << endl;
			qu.push(i);
		}
	});

	thread t2([&]() {
		for (int i = 0; i < 10; i++)
		{
			auto item = qu.front();
			qu.pop();
			cout << "consumed " << item << endl;
		}
	});

	t1.join();
	t2.join();

	return 0;
}

一些更有效的方法

ASYNC

#include<iostream>
#include<future>
#include<chrono>

using namespace std;

int work(int id)
{
	for (int i = 0; i < 5; i++) {
		cout << "running " << this_thread::get_id() << endl;
		this_thread::sleep_for(chrono::milliseconds(1000));
	}

	return id * 7;
}

int main()
{
	cout << "main id: " << this_thread::get_id() << endl;
	future<int> f1 = async(launch::async, work, 0);
	future<int> f2 = async(launch::async, work, 1); //会自动创建线程来处理
	cout << f1.get() << endl;
	cout << f2.get() << endl;

	return 0;
}

分布式工作

#include<iostream>
#include<thread>
#include<future>
#include<chrono>
#include<vector>
#include<mutex>
#include<iomanip>
#include<cmath>

using namespace std;

mutex g_mtx;

double calculate_pi(int terms, int start, int skip) {
	double res = 0.0;
	for (int i = start; i < terms; i += skip) {
		int sign = pow(-1, i);
		double term = 1.0 / (i * 2 + 1);
		res += sign * term;
	}
	return res * 4;
}

int main()
{
	vector<shared_future<double>> futures;

	const int CONCURRENCY = thread::hardware_concurrency();

	for (int i = 0; i < CONCURRENCY; i++) {
		shared_future<double> f = async(launch::async, calculate_pi, 1e9, i, CONCURRENCY);
		futures.push_back(f);
	}

	double sum = 0.0;
	for (auto f : futures) {
		sum += f.get();
	}

	// cout << setprecision(15) << "PI: " << M_PI << endl;
	cout << setprecision(15) << "Sum: " << sum << endl;

	return 0;
}
设计模式总结
从网线到加密：交换机、路由器、TC...

评论