C++消息队列的实现

Myth丶恋晨 2023-10-18 00:54 20阅读 0赞

（1）消息队列的实现

#ifndef NET_FRAME_CONCURRENT_QUEUE_H
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
          
          
           #define NET_FRAME_CONCURRENT_QUEUE_H
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
          
          
           #include <queue>
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
          
          
           #include <mutex>
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
          
          
           #include <condition_variable>
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
              
          
          
           template<
          
          
           class Type>
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
          
          
               /*消息队列实现*/
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
          
          
               class ConcurrentQueue {
            
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                  ConcurrentQueue& 
          
          
           operator=(
          
          
           const ConcurrentQueue&) = 
          
          
           delete;
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                  ConcurrentQueue(
          
          
           const ConcurrentQueue& other) = 
          
          
           delete;
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
              
          
          
           public:
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                  ConcurrentQueue() : _queue(), _mutex(), _condition() { }
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
                  
          
          
           virtual ~ConcurrentQueue() { }
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
                  
          
          
           void Push(Type record) {
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
                      
          
          
           std::lock_guard <
          
          
           std::mutex> lock(_mutex);
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                      _queue.push(record);
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                      _condition.notify_one();
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                  }
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
                  
          
          
           bool Pop(Type& record, bool isBlocked = true) {
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
                      
          
          
           if (isBlocked) {
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
                          
          
          
           std::unique_lock <
          
          
           std::mutex> lock(_mutex);
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
                          
          
          
           while (_queue.empty()) {
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                              _condition.wait(lock);
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                          }
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                      }
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
                      
          
          
           else 
          
          
           // If user wants to retrieve data in non-blocking mode
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                      {
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
                          
          
          
           std::lock_guard <
          
          
           std::mutex> lock(_mutex);
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
                          
          
          
           if (_queue.empty()) {
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
                              
          
          
           return 
          
          
           false;
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                          }
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                      }
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                      record = 
          
          
           std::move(_queue.front());
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                      _queue.pop();
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
                      
          
          
           return 
          
          
           true;
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                  }
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
                  
          
          
           int32_t Size() {
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
                      
          
          
           std::lock_guard <
          
          
           std::mutex> lock(_mutex);
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
                      
          
          
           return _queue.size();
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                  }
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
                  
          
          
           bool Empty() {
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
                      
          
          
           std::lock_guard <
          
          
           std::mutex> lock(_mutex);
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
                      
          
          
           return _queue.empty();
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                  }
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
              
          
          
           private:
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
                  
          
          
           std::
          
          
           queue <Type> _queue;
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
                  
          
          
           mutable 
          
          
           std::mutex _mutex;
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
                  
          
          
           std::condition_variable _condition;
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
              };
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
          
          
           #endif //NET_FRAME_CONCURRENT_QUEUE_H

（2）拥有消息队列的线程池的实现

.h文件如下

#ifndef NET_FRAME_THREAD_POOL_H
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
          #define NET_FRAME_THREAD_POOL_H
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
          #include "ConcurrentQueue.h"
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
          #include 
          
          
           <vector>
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
          #include 
          
          
           <queue>
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
          #include 
          
          
           <memory>
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
          #include 
          
          
           <thread>
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
          #include 
          
          
           <mutex>
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
          #include 
          
          
           <condition_variable>
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
          #include 
          
          
           <future>
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
          #include 
          
          
           <functional>
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
          #include 
          
          
           <stdexcept>
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
          #define MIN_THREADS 10
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
              template
          
          
           <class Type>
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
              class ThreadPool {
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                  ThreadPool& operator=(const ThreadPool&) = delete;
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                  ThreadPool(const ThreadPool& other) = delete;
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
              public:
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                  ThreadPool(int32_t threads, std::function
          
          
           <void(Type& record)> handler);
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                  virtual ~ThreadPool();
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                  void Submit(Type record);
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
              private:
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
              private:
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                  bool _shutdown;
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                  int32_t _threads;
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                  std::function
          
          
           <void(Type& record)> _handler;
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                  std::vector 
          
          
           <std::thread> _workers;
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                  ConcurrentQueue 
          
          
           <Type> _tasks;
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
              };
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
              template
          
          
           <class Type>
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
              ThreadPool
          
          
           <Type>::ThreadPool(int32_t threads, std::function
          
          
           <void(Type &record)> handler)
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                      : _shutdown(false),
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                        _threads(threads),
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                        _handler(handler),
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                        _workers(),
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                        _tasks() {
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                  if (_threads 
          
          
           < MIN_THREADS)
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
          
          
                       _threads = MIN_THREADS;
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
          
          
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
          
          
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
          
          
                   for (int32_t i = 0; i < _threads; ++i)
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
          
          
                       _workers.emplace_back(
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
          
          
                               [this] {
            
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
          
          
                                   while (!_shutdown) {
            
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
          
          
                                       Type record;
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
          
          
                                       _tasks.Pop(record, true);
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
          
          
                                       _handler(record);
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
          
          
                                   }
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
          
          
                               }
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
          
          
                       );
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
          
          
               }
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
          
          
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
          
          
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
          
          
               template<class Type>
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
              ThreadPool
          
          
           <Type>::~ThreadPool() {
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                  for (std::thread &worker: _workers)
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                      worker.join();
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
              }
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
              template
          
          
           <class Type>
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
              void ThreadPool
          
          
           <Type>::Submit(Type record) {
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
                  _tasks.Push(record);
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
              }
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
           
         
         
          
        
        
         
        
        
        
         
         
         
          
        
        
         
        
        
         
         
         
          
           
          
          #endif //NET_FRAME_THREAD_POOL_H