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

迈不过友情╰ 2023-01-02 02:28 255阅读 0赞

# 0.背景 #

测试多线程共享操作同一个队列, 当使用\#include<queue>原始的队列中queue<int>q时,没有发生问题,可以正常实现出入队. 当使用queue<char \*>q 时,出现了不稳定的数据重复和丢失...多线程编程需要实现一个线程安全的队列，上锁，避免多个线程同时读写.所以在网上找了个加了锁的队列,代码如第一部分.其中具体实现简例如第二部分.

# 1. 简单封装 #

在C++原有容器上面进行简单封装即可实现一个线程安全的队列，实现代码如下：[(参考)][Link 1]

#pragma once
    #include <atomic>
    #include <condition_variable>
    #include <functional>
    #include <memory>
    #include <mutex>
    #include <queue>
     
    template <class T>
    class Queue {
    protected:
    	// Data
    	std::queue<T> _queue;
    	typename std::queue<T>::size_type _size_max;
     
    	// Thread gubbins
    	std::mutex _mutex;
    	std::condition_variable _fullQue;
    	std::condition_variable _empty;
     
    	// Exit
    	// 原子操作
    	std::atomic_bool _quit; //{ false };
    	std::atomic_bool _finished; // { false };
     
    public:
    	Queue(const size_t size_max) :_size_max(size_max) {
    		_quit = ATOMIC_VAR_INIT(false);
    		_finished = ATOMIC_VAR_INIT(false);
    	}
     
    	bool push(T& data)
    	{
    		std::unique_lock<std::mutex> lock(_mutex);
    		while (!_quit && !_finished)
    		{
    			if (_queue.size() < _size_max)
    			{
    				_queue.push(std::move(data));
    				//_queue.push(data);
    				_empty.notify_all();
    				return true;
    			}
    			else
    			{
    				// wait的时候自动释放锁，如果wait到了会获取锁
    				_fullQue.wait(lock);
    			}
    		}
     
    		return false;
    	}
     
     
    	bool pop(T &data)
    	{
    		std::unique_lock<std::mutex> lock(_mutex);
    		while (!_quit)
    		{
    			if (!_queue.empty())
    			{
    				//data = std::move(_queue.front());
    				data = _queue.front();
    				_queue.pop();
     
    				_fullQue.notify_all();
    				return true;
    			}
    			else if (_queue.empty() && _finished)
    			{
    				return false;
    			}
    			else
    			{
    				_empty.wait(lock);
    			}
    		}
    		return false;
    	}
     
    	// The queue has finished accepting input
    	void finished()
    	{
    		_finished = true;
    		_empty.notify_all();
    	}
     
    	void quit()
    	{
    		_quit = true;
    		_empty.notify_all();
    		_fullQue.notify_all();
    	}
     
    	int length()
    	{
    		return static_cast<int>(_queue.size());
    	}
    };

# 2.简例 #

#include <iostream>
    
    #include <thread>
    
    #include <unistd.h>
    
    #include <atomic>
    
    #include <condition_variable>
    
    #include <functional>
    
    #include <memory>
    
    #include <mutex>
    
    #include <queue>
    
    
    
    template <class T>
    
    class Queue
    
    {
    
    protected:
    
        // Data
    
        std::queue<T> _queue;
    
        typename std::queue<T>::size_type _size_max;
    
    
    
        // Thread gubbins
    
        std::mutex _mutex;
    
        std::condition_variable _fullQue;
    
        std::condition_variable _empty;
    
    
    
        // Exit
    
        // 原子操作
    
        std::atomic_bool _quit;     //{ false };
    
        std::atomic_bool _finished; // { false };
    
    
    
    public:
    
        Queue(const size_t size_max) : _size_max(size_max)
    
        {
    
            _quit = ATOMIC_VAR_INIT(false);
    
            _finished = ATOMIC_VAR_INIT(false);
    
        }
    
    
    
        bool push(T &data)
    
        {
    
            std::unique_lock<std::mutex> lock(_mutex);
    
            while (!_quit && !_finished)
    
            {
    
                if (_queue.size() < _size_max)
    
                {
    
                    _queue.push(std::move(data));
    
                    //_queue.push(data);
    
                    _empty.notify_all();
    
                    return true;
    
                }
    
                else
    
                {
    
                    // wait的时候自动释放锁，如果wait到了会获取锁
    
                    _fullQue.wait(lock);
    
                }
    
            }
    
    
    
            return false;
    
        }
    
    
    
        bool pop(T &data)
    
        {
    
            std::unique_lock<std::mutex> lock(_mutex);
    
            while (!_quit)
    
            {
    
                if (!_queue.empty())
    
                {
    
                    //data = std::move(_queue.front());
    
                    data = _queue.front();
    
                    _queue.pop();
    
    
    
                    _fullQue.notify_all();
    
                    return true;
    
                }
    
                else if (_queue.empty() && _finished)
    
                {
    
                    return false;
    
                }
    
                else
    
                {
    
                    _empty.wait(lock);
    
                }
    
            }
    
            return false;
    
        }
    
    
    
        // The queue has finished accepting input
    
        void finished()
    
        {
    
            _finished = true;
    
            _empty.notify_all();
    
        }
    
    
    
        void quit()
    
        {
    
            _quit = true;
    
            _empty.notify_all();
    
            _fullQue.notify_all();
    
        }
    
    
    
        int length()
    
        {
    
            return static_cast<int>(_queue.size());
    
        }
    
    };
    
    
    
    Queue<int> q(1000);
    
    
    
    using namespace std;
    
    
    
    void t1() // 入队
    
    {
    
        while (1)
    
        {
    
            for (int i = 0; i <= 1000; i++)
    
            {
    
                sleep(0.1);
    
                q.push(i);
    
            }
    
        }
    
    }
    
    
    
    void t2() // 100ms 发送一次队首数据
    
    {
    
        while (1)
    
        {
    
            while (q.length() != 0)
    
            {
    
                sleep(0.1);
    
                int a;
    
                cout << "qLength: " << q.length() << " top:  " << a << endl;
    
                q.pop(a);
    
            }
    
        }
    
    }
    
    
    
    int main()
    
    {
    
        thread th1(t1);
    
        thread th2(t2);
    
        th1.join(); //等待th1执行完
    
        th2.join(); //等待th2执行完
    
        return 0;
    
    }
    
    // qLength: 4 top:  0
    
    // qLength: 4 top:  0
    
    // qLength: 4 top:  1
    
    // qLength: 3 top:  2
    
    // qLength: 4 top:  3
    
    // qLength: 6 top:  4
    
    // qLength: 5 top:  5
    
    // qLength: 5 top:  6
    
    // qLength: 4 top:  7

[Link 1]: https://blog.csdn.net/fengchu3415/article/details/82453808