利用条件变量构造线程安全队列

Posted on | In

简介

本文介绍如何使用条件变量控制并发的同步操作，试想有一个线程A一直输出1，另一个线程B一直输出2。我想让两个线程交替输出1，2，1，2…之类的效果，该如何实现？有的同学可能会说不是有互斥量mutex吗？可以用一个全局变量num表示应该哪个线程输出，比如num为1则线程A输出1，num为2则线程B输出2，mutex控制两个线程访问num，如果num和线程不匹配，就让该线程睡一会，这不就实现了吗？比如线程A加锁后发现当前num为2则表示它不能输出1，就解锁，将锁的使用权交给线程A，线程B就sleep一会。

不良实现

上面说的方式可以实现我们需要的功能，代码如下

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
void PoorImpleman() {
    std::thread t1([]() {
		for (;;) {

			{
				std::lock_guard<std::mutex> lock(mtx_num);
				if (num == 1) {
					std::cout << "thread A print 1....." << std::endl;
					num++;
					continue;
				}
			}

			std::this_thread::sleep_for(std::chrono::milliseconds(500));

        }

     });

	std::thread t2([]() {
		for (;;) {

			{
				std::lock_guard<std::mutex> lock(mtx_num);
				if (num == 2) {
					std::cout << "thread B print 2....." << std::endl;
					num--;
					continue;
				}
			}

			std::this_thread::sleep_for(std::chrono::milliseconds(500));

		}

		});

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

PoorImpleman虽然能实现我们交替打印的功能，会造成消息处理的不及时，因为线程A要循环检测num值，如果num不为1，则线程A就睡眠了，在线程A睡眠这段时间很可能B已经处理完打印了，此时A还在睡眠，是对资源的浪费，也错过了最佳的处理时机。所以我们提出了用条件变量来通知线程的机制，当线程A发现条件不满足时可以挂起，等待线程B通知，线程B通知A后，A被唤醒继续处理。

条件变量

我们这里用条件变量实现上面的逻辑

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
void ResonableImplemention() {
	std::thread t1([]() {
		for (;;) {
			
			std::unique_lock<std::mutex> lock(mtx_num);
			cvA.wait(lock, []() {
				return num == 1;
				});
			
			num++;
			std::cout << "thread A print 1....." << std::endl;
			cvB.notify_one();
		}

		});

	std::thread t2([]() {
		for (;;) {

			std::unique_lock<std::mutex> lock(mtx_num);
			cvB.wait(lock, []() {
				return num == 2;
				});

			num--;
			std::cout << "thread B print 2....." << std::endl;
			cvA.notify_one();
		}

		});

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

当条件不满足时(num 不等于1 时)cvA.wait就会挂起，等待线程B通知通知线程A唤醒，线程B采用的是cvA.notifyone。
这么做的好处就是线程交替处理非常及时。比起sleep的方式，我们可以从控制台上看出差异效果，sleep的方式看出日志基本是每隔1秒才打印一次，效率不高。

线程安全队列

之前我们实现过线程安全的栈，对于pop操作，我们如果在线程中调用empty判断是否为空，如果不为空，则pop，因为empty和pop内部分别加锁，是两个原子操作，导致pop时可能会因为其他线程提前pop导致队列为空，从而引发崩溃。我们当时的处理方式是实现了两个版本的pop，一种是返回智能指针类型，一种通过参数为引用的方式返回。对于智能指针版本我们发现队列为空则返回空指针，对于引用版本，
发现队列为空则抛出异常，这么做并不是很友好，所以我们可以通过条件变量完善之前的程序，不过这次我们重新实现一个线程安全队列。

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
#include <queue>
#include <mutex>
#include <condition_variable>
template<typename T>
class threadsafe_queue
{
private:
    std::mutex mut;
    std::queue<T> data_queue;
    std::condition_variable data_cond;
public:
    void push(T new_value)
    {
        std::lock_guard<std::mutex> lk(mut);
        data_queue.push(new_value);
        data_cond.notify_one();
    }
    void wait_and_pop(T& value)
    {
        std::unique_lock<std::mutex> lk(mut);
        data_cond.wait(lk,[this]{return !data_queue.empty();});
        value=data_queue.front();
        data_queue.pop();
    }
};

threadsafe_queue<data_chunk> data_queue;    
void data_preparation_thread()
{
    while(more_data_to_prepare())
    {
        data_chunk const data=prepare_data();
        data_queue.push(data);    ⇽---  ②
    }
}
void data_processing_thread()
{
    while(true)
    {
        data_chunk data;
        data_queue.wait_and_pop(data);   
        process(data);
        if(is_last_chunk(data))
            break;
    }
}

我们可以启动三个线程，一个producer线程用来向队列中放入数据。一个consumer1线程用来阻塞等待pop队列中的元素。

另一个consumer2尝试从队列中pop元素，如果队列为空则直接返回，如果非空则pop元素。

打印时为了保证线程输出在屏幕上不会乱掉，所以加了锁保证互斥输出

测试代码如下

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
void test_safe_que() {
	threadsafe_queue<int>  safe_que;
	std::mutex  mtx_print;
 	std::thread producer(
		[&]() {
			for (int i = 0; ;i++) {
				safe_que.push(i);
				{
					std::lock_guard<std::mutex> printlk(mtx_print);
					std::cout << "producer push data is " << i << std::endl;
				}
				
				std::this_thread::sleep_for(std::chrono::milliseconds(200));
			}
		}
	);

	std::thread consumer1(
		[&]() {
			for (;;) {
				auto data = safe_que.wait_and_pop();
				{
					std::lock_guard<std::mutex> printlk(mtx_print);
					std::cout << "consumer1 wait and pop data is " << *data << std::endl;
				}
				
				std::this_thread::sleep_for(std::chrono::milliseconds(500));
			}
		}	
	);

	std::thread consumer2(
		[&]() {
			for (;;) {
				auto data = safe_que.try_pop();
				if (data != nullptr) {
					{
						std::lock_guard<std::mutex> printlk(mtx_print);
						std::cout << "consumer2 try_pop data is " << *data << std::endl;
					}
					
				}

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

	producer.join();
	consumer1.join();
	consumer2.join();
}

测试效果如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
producer push data is 0
consumer1 wait and pop data is 0
producer push data is 1
producer push data is 2
consumer2 try_pop data is 1
consumer1 wait and pop data is 2
producer push data is 3
producer push data is 4
consumer2 try_pop data is 3
consumer1 wait and pop data is 4
producer push data is 5
producer push data is 6
producer push data is 7
consumer2 try_pop data is 5
consumer1 wait and pop data is 6

我们能看到consumer1和consumer2是并发消费的

总结

本文介绍了如何通过条件变量实现并发线程的同步处理。

视频链接

https://space.bilibili.com/271469206/channel/collectiondetail?sid=1623290

源码链接

https://gitee.com/secondtonone1/boostasio-learn