【多线程学习】线程池、简单的银行取款模拟

学习代码如下，教程来自：http://www.seestudy.cn/?list_9/42.html

#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
// 1
void printFun(std::string msg) {
	std::cout << msg << std::endl;
	//for (int i = 0; i < 1000; i++) {
	//	std::cout << i;
	//}
	return;
}

void printHelloWorld() {
	std::cout << "Hello World" << std::endl;
	return;
}

// 2
void foo(int& x) {
	x += 1;
}

// 3、线程安全和互斥锁
int a = 0;
std::mutex mtx;
void func() {
	for (int i = 0; i < 100000; i++) {
		mtx.lock();
		a += 1;
		mtx.unlock();
	}
}

// 4、互斥死锁
int c = 0;
std::mutex mtx1;
std::mutex mtx2;
void func1() {
	for (int i = 0; i < 400; i++) {
		mtx1.lock();
		mtx2.lock();
		c += 1;
		mtx1.unlock();
		mtx2.unlock();
	}
}

void func2() {
	for (int i = 0; i < 400; i++) {
		mtx1.lock();
		mtx2.lock();
		c += 1;
		mtx1.unlock();
		mtx2.unlock();
	}
}

// 5
std::timed_mutex mtx3;
int d = 0;
void func3() {
	for (int i = 0; i < 2; i++) {
		//std::lock_guard lg(mtx3);
		std::unique_lock<std::timed_mutex> lg(mtx3, std::defer_lock);
		if (lg.try_lock_for(std::chrono::seconds(2))) {
			std::this_thread::sleep_for(std::chrono::seconds(1));
			d += 1;
		}
		
	}
}

// 6 单例模式
class Log {
private:
	Log() {};//单例模式要保证构造函数私有，然后在类里面定义静态函数得到一个属于整个类的实例
	//static Log* log;
	//static std::once_flag initFlag;  // once_flag 对象
public:

	Log(const Log& log) = delete;
	Log& operator=(const Log& log) = delete;

	static Log& GetInstance() {
		//饿汉模式，线程安全
		static Log log;
		return log;

		//懒汉模式，多线程必须加锁
		//if (!log) log = new Log;
		//std::call_once(initFlag, []() {  // 使用 lambda 函数创建单例实例
		//	log = new Log;
		//	});
		//return *log;
	}

	void printLog(std::string msg) {
		std::cout << __TIME__ << " " << msg << std::endl;
	}
};
//Log* Log::log = nullptr;
//std::once_flag Log::initFlag;

void func4() {
	Log::GetInstance().printLog("error");
}

// 7、生产者消费者模型
std::queue<int> g_queue;
std::condition_variable g_cv;
std::mutex mtx_produce;
void Producer() { //生产者
	for (int i = 0; i < 10; i++) {
		std::unique_lock<std::mutex> lock(mtx_produce);
		g_queue.push(i);
		//通知消费者取任务
		g_cv.notify_one();
		std::cout << "Producer : " << i << std::endl;

		std::this_thread::sleep_for(std::chrono::microseconds(500));
	}
	
}

void Consumer() { //消费者
	while (1) {
		std::unique_lock<std::mutex> lock(mtx_produce);
		g_cv.wait(lock, []() {return !g_queue.empty(); });
		int value = g_queue.front();
		g_queue.pop();
		std::cout << "Consumer : " << value << std::endl;
	}
}

// 8、线程池
class ThreadPool{
public:
	ThreadPool(int numThreads) :stop(false) { //构造函数
		for (int i = 0; i < numThreads; i++) {
			threads.emplace_back([this] { //每个线程执行的操作
				while (1) {
					std::unique_lock<std::mutex> lock(mtxPool); //1、加锁
					condition.wait(lock, [this] { //2、将线程置于休眠状态，直到有任务可用或线程池停止
						return !tasks.empty() || stop; //如果有新任务了或者线程池要关闭了，则退出休眠状态
						});

					if (stop && tasks.empty()) { //如果线程池关闭且任务队列为空，直接返回
						return;
					}

					std::function<void()> task(std::move(tasks.front())); //3、从任务队列中取任务，std::move减少拷贝，移动语义直接转移资源
					tasks.pop(); 
					lock.unlock(); //4、解锁
					task();
				}
			} //每个线程操作结束
			);
		}
	}

	~ThreadPool() { //析构函数
		{
			std::unique_lock<std::mutex> lock(mtxPool);
			stop = true; 
		}
		
		condition.notify_all(); //通知所有线程要关闭了，但线程不会立即停止，它们会在检查stop变量后做出决策
		for (auto& t : threads) {
			t.join(); //等待所有线程执行完
		}
	}

	template<class F, class... Args>
	void enqueue(F&& f, Args&&... args) { //将任务封装成std::function 对象，并添加到任务队列中
		std::function<void()> task = std::bind(std::forward<F>(f), std::forward<Args>(args)...);
		{
			std::unique_lock<std::mutex> lock(mtxPool);
			tasks.emplace(std::move(task)); 
		}

		condition.notify_one(); //通知一个休眠的线程有新的任务可以取
	};
	

private:
	std::vector<std::thread> threads; //线程数组
	std::queue<std::function<void()>> tasks; //任务队列
	std::mutex mtxPool; //互斥量
	std::condition_variable condition; //条件变量
	bool stop; //线程池停止的标记
};


int main() {
	
	//1、创建线程
	std::thread thread1(printFun, "Hello Thread");
	bool isJoin = thread1.joinable();
	if (isJoin) {
		thread1.join();
	}
	std::cout << "over" << std::endl;
	std::thread thread2(printHelloWorld);
	thread2.join();

	int b = 1;
	std::thread thread3(foo, std::ref(b));
	thread3.join();

	//3、线程冲突
	std::thread t1(func);
	std::thread t2(func);
	t1.join();
	t2.join();
	std::cout << a << std::endl;

	//4、互斥量死锁

	std::thread t3(func1);
	std::thread t4(func2);
	t3.join();
	t4.join();
	std::cout << c << std::endl;

	//5、lock_guard 与 std::unique_lock
	std::thread t5(func3);
	std::thread t6(func3);
	t5.join();
	t6.join();
	std::cout << d << std::endl;

	//6、单例模式
	//Log a1; 该行会报错，因为单例模式只能有一个实例，就是在类里面
	std::thread t7(func4);
	std::thread t8(func4);
	t7.join();	
	t8.join();

	//7、生产者与消费者模型
	std::thread t9(Producer);
	std::thread t10(Consumer);
	t9.join();
	t10.join();
	

	//8、跨平台线程池
	ThreadPool pool(4);
	for (int i = 0; i < 10; i++) {
		pool.enqueue([i] { //加任务
			std::cout << "task :" << i << " is running " << std::endl;
			std::this_thread::sleep_for(std::chrono::seconds(1));
			std::cout << "task :" << i << " is done " << std::endl;
			});
	}
	return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261

线程池解读：

在示例中，我们定义了一个 ThreadPool
类，并且在构造函数中创建了指定数目的线程。在每个线程中，我们不断地从任务队列中获取任务并执行，直到线程池被停止。在 enqueue()
函数中，我们将任务封装成一个 std::function 对象，并将它添加到任务队列中。在 ThreadPool
的析构函数中，我们等待所有线程执行完成后再停止所有线程。

在主函数中，我们创建了一个 ThreadPool 对象，并向任务队列中添加了 8 个任务。每个任务会输出一些信息，并且在执行完后等待 1
秒钟。由于线程池中有 4 个线程，因此这 8 个任务会被分配到不同的线程中执行。在任务执行完成后，程序会退出。

对于多线程enqueue函数的解读：
该函数作用是将用户传入的函数（以及其参数）封装为一个可以无参数调用的任务，并将这个任务添加到任务队列中。下面逐行解释代码：

template：模板参数声明。F用于匹配用户传入的函数类型，Args…是一个可变模板参数，用于匹配用户传入的函数参数。

void enqueue(F&& f, Args&&... args)：enqueue函数的定义，其中使用了右值引用（&&）来接受传入的函数和其参数。这样可以有效地处理传入的临时对象和实现完美转发。

std::function task = std::bind(std::forward(f), std::forward(args)...);：使用std::bind将用户传入的函数和其参数绑定起来，创建一个可以无参数调用的函数对象。
std::forward(f)和std::forward(args)...是完美转发的技巧，它确保传入enqueue的参数以其原始形式（lvalue或rvalue）传递给std::bind。
将绑定后的函数对象赋给std::function类型的task，这样可以统一地处理任务，不管用户传入的函数有什么签名。
{ 一个局部代码块的开始。确保std::unique_lock的作用范围仅限于这个块，这样当块结束时，锁会自动释放。
std::unique_lock lock(mtxPool);：使用std::unique_lock锁住mtxPool互斥量，确保在修改任务队列时不会有其他线程同时修改。

tasks.emplace(std::move(task));：将task移动到任务队列中。使用std::move是为了避免不必要的拷贝，而是将task的资源转移到队列中的新任务对象。

} 局部代码块结束，这时std::unique_lock会自动释放锁。

condition.notify_one();：使用条件变量condition通知一个正在等待的线程：任务队列中有了新的任务，可以进行处理了。

通过这个函数，用户可以很方便地将任意函数（及其参数）提交给线程池执行。

自己写的一个模拟银行窗口取钱的小程序，目的是为了更好地理解和应用多线程。
比较简单的一版，用了std::mutex mtx; mtx.lock()手动上锁和mtx.unlock()解锁：

#include 
#include 
#include 
using namespace std;

int sumMoney = 10000;
mutex mtx;

void drawMoney(string window, int num) {
	while (sumMoney > 0) {
		mtx.lock();
		if (sumMoney >= num) {
			sumMoney -= num;
			cout << window << "取 " << num << " 元钱，现金储备还剩：" << sumMoney << endl;
		}
		mtx.unlock();
		this_thread::sleep_for(chrono::microseconds(100)); //保证每个窗口都有概率取到钱
	}
}

int main() {
	thread t1(drawMoney, "银行窗口1", 100);
	thread t2(drawMoney, "银行窗口2", 200);
	thread t3(drawMoney, "银行窗口3", 500);
	t1.join();
	t2.join();
	t3.join();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

第二版想试试condition_variable，但实际上没有必要用，因为条件变量主要用于复杂的同步场景，例如当某个线程需要等待特定条件满足时（例如任务队列中有任务可供处理），本问题
这个简单场景中核心在于线程间的取款操作同步，而不是等待特定的条件发生。因此，使用std::mutex就足够了。通过适当地加锁、解锁以及合理地使用线程调度提示（如this_thread::yield()），可以使多个线程有均匀的取款机会。

#include 
#include 
#include 
#include 
using namespace std;

int sumMoney = 10000;
mutex mtx;
condition_variable cv;

void drawMoney(string window, int num) {
    unique_lock<mutex> lock(mtx);

    while (sumMoney > 0) {
        while (sumMoney < num) {  // 如果不够这个线程取，就等待
            cv.wait(lock);
        }

        sumMoney -= num;
        cout << window << "取 " << num << " 元钱，现金储备还剩：" << sumMoney << endl;
        cv.notify_all();  // 通知所有线程，因为余额发生了变化
    }
}

int main() {
    thread t1(drawMoney, "银行窗口1", 100);
    thread t2(drawMoney, "银行窗口2", 200);
    thread t3(drawMoney, "银行窗口3", 500);
    t1.join();
    t2.join();
    t3.join();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32