c++11 多线程编程——如何实现线程安全队列

时间:2022-11-25 16:14:09

线程安全队列的接口文件如下:

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
#include <memory>
template<typename T>
class threadsafe_queue {
 public:
  threadsafe_queue();
  threadsafe_queue(const threadsafe_queue&);
  threadsafe_queue& operator=(const threadsafe_queue&) = delete;
 
  void push(T new_value);
 
  bool try_pop(T& value);
  std::shared_ptr<T> try_pop();
 
  void wait_and_pop(T& value);
  std::shared_ptr<T> wait_and_pop();
 
  bool empty() const;
};

push函数

push()函数实现向队列添加数据的功能。添加数据后,使用std::condition_variable的notify_one通知取数据时被阻塞的线程。

?
1
2
3
4
5
6
void push(T tValue) {
  std::shared_ptr<T> data(std::make_shared<T>(std::move(tValue)));
  std::lock_guard<std::mutex> lk(mut);
  data_queue.push(data);
  data_con.notify_one();
}

wait_and_pop函数

wait_and_pop()函数实现从队列取数据的功能,当队列为空时,线程被挂起,等待有数据时被唤醒。

注意,这两个函数中没有使用std::lock_guard,而是使用std::unique_lock,这是为什么呢?

这是因为std::condition_variable的wait函数会首先检测条件data_queue.empty()是否满足,如果队列为空,wait函数会释放mutex,并被挂起;当有新的数据进入队列,std::condition_variable的wait函数会被唤醒,重新尝试获取mutex,然后检测队列是否为空,如果队列非空,则继续向下执行。由于函数的执行过程存在锁的释放和重新获取,所以没有使用std::lock_guard,而是选择了std::unique_lock。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
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();
}
 
std::shared_ptr<T> wait_and_pop() {
  std::unique_lock<std::mutex> lk(mut);
  data_cond.wait(lk,[this]{return !data_queue.empty();});
  std::shared_ptr<T> res(std::make_shared<T>(data_queue.front()));
  data_queue.pop();
  return res;
}

try_pop函数

try_pop函数提供非阻塞调用下的弹出队列(queue)的功能。弹出成功返回true或者非空shared_ptr,失败则返回false或者nullptr。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
bool try_pop(T& value) {
  std::lock_guard<std::mutex> lk(mut);
  if(data_queue.empty()) {
    return false;
  }
  value = data_queue.front();
  data_queue.pop();
  
  return true;
}
 
std::shared_ptr<T> try_pop() {
  std::lock_guard<std::mutex> lk(mut);
  if(data_queue.empty()) {
    return std::shared_ptr<T>();
  }
  std::shared_ptr<T> res(std::make_shared<T>(data_queue.front()));
  data_queue.pop();
  return res;
}

empty函数

?
1
2
3
4
bool empty() const {
  std::lock_guard<std::mutex> lk(mut);
  return data_queue.empty();
}

这里注意,empty()是const类型的成员函数,表明它声明自己并不改变任何成员变量,但是mutex lock是一个mutating opertation,所以必须要将mut声明为mutable类型(mutable std::mutex mut)。

完整代码如下:

?
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
#include <queue>
#include <memory>
#include <mutex>
#include <condition_variable>
 
template<typename T>
class threadsafe_queue {
 private:
   mutable std::mutex mut;
   std::queue<T> data_queue;
   std::condition_variable data_cond;
 public:
   threadsafe_queue(){}
   threadsafe_queue(threadsafe_queue const& other) {
     std::lock_guard<std::mutex> lk(other.mut);
     data_queue=other.data_queue;
   }
 
   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();
   }
 
   std::shared_ptr<T> wait_and_pop() {
     std::unique_lock<std::mutex> lk(mut);
     data_cond.wait(lk,[this]{return !data_queue.empty();});
     std::shared_ptr<T> res(std::make_shared<T>(data_queue.front()));
     data_queue.pop();
     return res;
   }
 
   bool try_pop(T& value) {
     std::lock_guard<std::mutex> lk(mut);
     if(data_queue.empty())
       return false;
     value=data_queue.front();
     data_queue.pop();
     return true;
   }
 
   std::shared_ptr<T> try_pop() {
     std::lock_guard<std::mutex> lk(mut);
     if(data_queue.empty())
       return std::shared_ptr<T>();
     std::shared_ptr<T> res(std::make_shared<T>(data_queue.front()));
     data_queue.pop();
     return res;
   }
 
   bool empty() const {
     std::lock_guard<std::mutex> lk(mut);
     return data_queue.empty();
   }
};

以上就是c++ 如何实现线程安全队列的详细内容,更多关于c++ 线程安全队列的资料请关注服务器之家其它相关文章!