Muduo之封装BlockingQueue、BoundedBlockingQueue

比眉伴天荒 2022-05-31 05:30 148阅读 0赞

# 阻塞队列介绍 #

> 阻塞队列首先是一个队列，数据可以由队列的一端输入，从另外一端输出。多线程环境中，通过队列可以很容易实现数据共享，比如经典的“生产者”和“消费者”模型。假设我们有若干生产者线程，另外又有若干个消费者线程。如果生产者线程需要把准备好的数据共享给消费者线程，利用队列的方式来传递数据，就可以很方便地解决他们之间的数据共享问题。但如果生产者和消费者在某个时间段内，万一发生数据处理速度不匹配的情况呢？理想情况下，如果生产者产出数据的速度大于消费者消费的速度，并且当生产出来的数据累积到一定程度的时候，那么生产者必须暂停等待一下（阻塞生产者线程），以便等待消费者线程把累积的数据处理完毕，反之如果消费者太快，导致队列为空，那么消费者必须等待下（阻塞消费者线程）。（在多线程领域：所谓阻塞，在某些情况下会挂起线程（即阻塞），一旦条件满足，被挂起的线程又会自动被唤醒），下图是两种需要被阻塞的情况。  
> ![这里写图片描述][70]

# 2、BlockingQueue #

无界阻塞队列，就是生产者不会阻塞，不停的生产，仅仅出现队列为空阻塞消费者的情形。一个mutex和一个conditon即可搞定。下面是实现代码。

template<typename T>
    class BlockingQueue : boost::noncopyable
    {
     public:
      BlockingQueue()
        : mutex_(),//先构造初始化mutex_
          notEmpty_(mutex_),//再构造初始化Condition
          queue_()//构造队列
      {
      }
    
      void put(const T& x)
      {
        MutexLockGuard lock(mutex_);
        queue_.push_back(x);
        notEmpty_.notify(); //每次都通知消费者，队列非空，这里并不阻塞生产者。
      }
    
    #ifdef __GXX_EXPERIMENTAL_CXX0X__
      void put(T&& x)
      {
        MutexLockGuard lock(mutex_);
        queue_.push_back(std::move(x));//蹇界暐閲婃斁鏀炬弧
        notEmpty_.notify();
      }
      // FIXME: emplace()
    #endif
    
      T take()
      {
        MutexLockGuard lock(mutex_);
        // always use a while-loop, due to spurious wakeup
        while (queue_.empty()){
          notEmpty_.wait();//条件变量标准操作，循环测试，防止虚假唤醒，队列空，消费者阻塞
        }
        assert(!queue_.empty());//断定非空，用于系统测试
    #ifdef __GXX_EXPERIMENTAL_CXX0X__
        T front(std::move(queue_.front()));
    #else
        T front(queue_.front());//构造对象，值初始化
        queue_.pop_front();//弹出队列前面元素
        return front;//返回对象
      }
    
      size_t size() const
      {
        MutexLockGuard lock(mutex_);
        return queue_.size();
      }
    
     private:
      mutable MutexLock mutex_;//互斥量
      Condition         notEmpty_;//用于消费者等待通知条件 
      std::deque<T>     queue_;//数据队列
    };

测试代码：

# 3、BoundedBlockingQueue #

有界阻塞队列，队列空阻塞消费者，队列满阻塞生产者。1个mutex和2个conditon即可搞定。下面是实现代码。

template<typename T>
    class BoundedBlockingQueue : boost::noncopyable
    {
     public:
      explicit BoundedBlockingQueue(int maxSize)
        : mutex_(),//构造互斥锁
          notEmpty_(mutex_),//构造条件变量
          notFull_(mutex_),//构造条件变量
          queue_(maxSize)//初始化环形队列大小
      {
      }
    
      void put(const T& x)
      {
        MutexLockGuard lock(mutex_);
        while (queue_.full()){
       //缓冲区满，阻塞生产者
          notFull_.wait();
        }
        assert(!queue_.full());
        queue_.push_back(x);
        notEmpty_.notify();//缓冲区有数据，唤醒一个消费者
      }
    
      T take()
      {
        MutexLockGuard lock(mutex_);
        while (queue_.empty()){
       //缓冲区空，阻塞消费者
          notEmpty_.wait();
        }
        assert(!queue_.empty());
        T front(queue_.front());
        queue_.pop_front();
        notFull_.notify();//唤醒一个阻塞的生产者
        return front;//返回对象
      }
    
      bool empty() const
      {
        MutexLockGuard lock(mutex_);
        return queue_.empty();
      }
    
      bool full() const
      {
        MutexLockGuard lock(mutex_);
        return queue_.full();
      }
    
      size_t size() const
      {
        MutexLockGuard lock(mutex_);
        return queue_.size();
      }
    
      size_t capacity() const
      {
        MutexLockGuard lock(mutex_);
        return queue_.capacity();
      }
    
     private:
      mutable MutexLock          mutex_;//一个互斥锁
      Condition                  notEmpty_;//消费者条件变量
      Condition                  notFull_;//生产者条件变量
      boost::circular_buffer<T>  queue_;//环形缓冲区
    }

测试代码

[70]: /images/20220531/dc175c5a36c64826ac379f7225ea3912.png