C++实现线程安全的队列

一时失言乱红尘 2022-05-13 12:46 198阅读 0赞

C++标准库已经提供了std::queue这一队列容器，但不是线程安全的。std::queue这个容器已经提供了pop(),push(),empty()等这些读写操作容器的函数，只要在这些函数上面加个锁，就可以使其线程安全。  
  在C++原有容器上面进行简单封装即可实现一个线程安全的队列，实现代码如下：

#include <iostream>
    #include <string>
    #include <condition_variable>
    #include <mutex>
    #include <queue>
    #include <memory>
    
    template<class T, class Container = std::queue<T>>
    class ThreadSafeQueue {
    public:
      ThreadSafeQueue() = default;
    
      template <class Element>
      void Push(Element&& element) {
        std::lock_guard<std::mutex> lock(mutex_);
        queue_.push(std::forward<Element>(element));
        not_empty_cv_.notify_one();
      }
    
      void WaitAndPop(T& t) {
        std::unique_lock<std::mutex> lock(mutex_);
        not_empty_cv_.wait(lock, []() {
          return !queue_.empty();
        });
    
        t = std::move(queue_.front());
        queue_.pop()
      }
    
      std::shared_ptr<T> WaitAndPop() {
        std::unique_lock<std::mutex> lock(mutex_);
        not_empty_cv_.wait(lock, [this]() {
          return !queue_.empty();
        });
    
        std::shared_ptr<T> t_ptr = std::make_shared<T>(queue_.front());
        queue_.pop();
    
        return t_ptr;
      }
    
      bool TryPop(T& t) {
        std::lock_guard<std::mutex> lock(mutex_);
        if (queue_.empty()) {
          return false;
        }
    
        t = std::move(queue_.front());
        queue_.pop()
    
          return true;
      }
    
      std::shared_ptr<T> TryPop() {
        std::lock_guard<std::mutex> lock(mutex_);
        if (queue_.empty()) {
          return std::shared_ptr<T>();
        }
    
        t = std::move(queue_.front());
        std::shared_ptr<T> t_ptr = std::make_shared<T>(queue_.front());
        queue_.pop();
    
        return t_ptr;
      }
    
      bool IsEmpty() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return queue_.empty();
      }
    
    private:
      ThreadSafeQueue(const ThreadSafeQueue&) = delete;
      ThreadSafeQueue& operator=(const ThreadSafeQueue&) = delete;
      ThreadSafeQueue(ThreadSafeQueue&&) = delete;
      ThreadSafeQueue& operator=(ThreadSafeQueue&&) = delete;
    
    private:
      Container queue_;
    
      std::condition_variable not_empty_cv_;
      mutable std::mutex mutex_;
    };

### 代码分析： ###

#### 1. 条件变量std::condition\_variable的使用。 ####

使用条件变量的原因是为了实现**WaitAndPop()**这个函数。这个函数的作用是如果容器中有数据则进行Pop，如果没有数据则进行等待。  
每次在Pop数据的时候都会调用条件变量 `not_empty_cv_.wait(）;`，如果满足wait()的条件则程序阻塞到这里，等待其他线程Push数据之后进行`not_empty_cv_.notify_one();`来唤醒wait()。

#### 2. 条件变量wait函数的作用。 ####

void WaitAndPop(T& t) {
        std::unique_lock<std::mutex> lock(mutex_);
        not_empty_cv_.wait(lock, []() {
          return !queue_.empty();
        });
    
        t = std::move(queue_.front());
        queue_.pop()
      }

在上述代码中，`wait()`之前首先会使用std::unique\_lock获取互斥元mutex\_，然后当代码阻塞到wait()这里的时候，Push函数也要获取互斥元mutex\_才能插入数据，这里看起来十分奇怪。实际上`wait()`函数将线程阻塞到这里的时候，会**解锁互斥元**，所以其他线程仍然可以正常获取mutex\_。当其他线程进行`notify_one()`的时候，会唤醒刚才阻塞等待的线程，该线程会重新上锁，然后判断跳出wait()的条件是否满足，如果满足则跳出wait(),进行后面操作，否则继续进行阻塞等待的动作。

#### 3. 使用wait()时需要传入锁std::unique\_lock。 ####

给条件变量wait()函数传入的锁是`std::unique_lock`，而不是`std::lock_guard`。因为线程在执行wait()函数的时候，如果进入等待，就要解锁，被唤醒后又会重新加锁，std::lock\_guard功能比较简单，不能满足wait()的要求，所以要用std::unique\_lock。

#### 4. 使用wait()时需要传入判断条件，来防止假唤醒。 ####

std::unique_lock<std::mutex> lock(mutex_);
        not_empty_cv_.wait(lock, []() {
          return !queue_.empty();
        })

这里的判断条件是lambda表达式`[](){ return !queue_.empty(); }`这个匿名函数就是用来判断队列是否不为空。这个条件判断其实就是退出等待的条件。`wait(lock, 退出等待条件函数)`，这条语句实际就是英文的表达习惯`wait until ...`，意思是`线程进行等待直到满足退出等待条件为止`。  
  如果这里没有加判断条件而是直接调用`wait(lock);`，这样会导致两个严重的问题：

1.  假唤醒问题，这种问题导致容器中还没有数据就进行了Pop。
2.  如果容器中已经有了数据，在Pop的时候还要等wait()收到notify才能Pop，这样导致了即使有数据也不能取出的问题。

#### 5. 成员变量std::mutex要用mutable修饰。 ####

mutable std::mutex mutex_;

`mutable`的作用是突破const的限制，使得一个成员函数在const修饰的时候，依然可以更改mutable修饰的成员变量。因为成员变量mutex\_，会不断地加锁解锁，所以互斥元必须是可变的。

#### 6. 使用引用折叠来简化代码。 ####

template <class Element>
      void Push(Element&& element) {
        std::lock_guard<std::mutex> lock(mutex_);
        queue_.push(std::forward<Element>(element));
        not_empty_cv_.notify_one();
      }

这段代码可以替换为：

void Push(const T& t) {
      std::lock_guard<std::mutex> lock(mutex_);
      queue_.push(t);
      not_empty_cv_.notify_one();
    }
    
    void Push(T&& t) {
      std::lock_guard<std::mutex> lock(mutex_);
      queue_.push(std::move(t));
      not_empty_cv_.notify_one();
    }