C++自用小轮子——线程安全队列

简介

记录开发时自用的小轮子：线程安全队列

线程安全队列

#ifndef THREADSAFEQUEUE_H
#define THREADSAFEQUEUE_H

#include <iostream>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <memory>

template<typename T>
class ThreadSafeQueue
{
public:
    ThreadSafeQueue();
    ThreadSafeQueue(const ThreadSafeQueue& other);
    ThreadSafeQueue& operator=(const ThreadSafeQueue& other);
    ~ThreadSafeQueue();
public:
    void push(T new_value);

    //对于pop方法，std::queue中的pop只负责弹出元素，不返回元素
    //这里为了接口简化，设计为在pop的同时返回弹出的元素
    //timeout为超时时间，单位为毫秒
    //默认阻塞pop
    bool wait_and_pop(T& value, std::chrono::milliseconds timeout_millisecond = std::chrono::milliseconds::max());
    std::shared_ptr<T> wait_and_pop(std::chrono::milliseconds timeout_millisecond = std::chrono::milliseconds::max());
    
    bool empty() const;
    size_t size() const;

    T& front();
    const T& front() const;
    std::shared_ptr<T> front_ptr();

private:
    mutable std::mutex mtx_;
    std::queue<T> data_queue_;
    std::condition_variable cv_data_; 
};

template<typename T>
ThreadSafeQueue<T>::ThreadSafeQueue()
{
}

template<typename T>
ThreadSafeQueue<T>::~ThreadSafeQueue()
{
}

template<typename T>
ThreadSafeQueue<T>::ThreadSafeQueue(const ThreadSafeQueue& other)
{
    std::lock_guard<std::mutex> lock(other.mtx_);
    data_queue_ = other.data_queue_;
}

template<typename T>
ThreadSafeQueue<T>& ThreadSafeQueue<T>::operator=(const ThreadSafeQueue& other)
{
    if (this == &other)
    {
        return *this;
    }
    //同时锁定两个互斥量，防止死锁
    std::lock(mtx_, other.mtx_);
    //领养锁，因为已经锁定了
    std::lock_guard<std::mutex> self_lock(mtx_, std::adopt_lock);
    std::lock_guard<std::mutex> other_lock(other.mtx_, std::adopt_lock);
    data_queue_ = other.data_queue_;
    return *this;
}

template<typename T>
void ThreadSafeQueue<T>::push(T new_value)
{
    std::lock_guard<std::mutex> lock(mtx_);
    data_queue_.push(std::move(new_value));
    cv_data_.notify_one();
}

template<typename T>
bool ThreadSafeQueue<T>::wait_and_pop(T& value, std::chrono::milliseconds timeout_millisecond)
{
    std::unique_lock<std::mutex> lock(mtx_);
    if (data_queue_.empty())
    {
        if (timeout_millisecond == std::chrono::milliseconds::max())
        {
            cv_data_.wait(lock, [this] {return !data_queue_.empty();});
        }
        else
        {
            if (!cv_data_.wait_for(lock, timeout_millisecond, [this] {return !data_queue_.empty();}))
                return false;
        }
    }
    value = std::move(data_queue_.front());
    data_queue_.pop();
    return true;
}

template<typename T>
std::shared_ptr<T> ThreadSafeQueue<T>::wait_and_pop(std::chrono::milliseconds timeout_millisecond)
{
    std::unique_lock<std::mutex> lock(mtx_);
    if (data_queue_.empty())
    {
        if (timeout_millisecond == std::chrono::milliseconds::max())
        {
            cv_data_.wait(lock, [this] {return !data_queue_.empty();});
        }
        else
        {
            if (!(cv_data_.wait_for(lock, timeout_millisecond, [this] {return !data_queue_.empty();})))
                return std::shared_ptr<T>();
        }
    }
    std::shared_ptr<T> res(std::make_shared<T>(std::move(data_queue_.front())));
    data_queue_.pop();
    return res;
}

template<typename T>
bool ThreadSafeQueue<T>::empty() const
{
    std::lock_guard<std::mutex> lock(mtx_);
    return data_queue_.empty();
}

template<typename T>
size_t ThreadSafeQueue<T>::size() const
{
    std::lock_guard<std::mutex> lock(mtx_);
    return data_queue_.size();
}

template<typename T>
T& ThreadSafeQueue<T>::front()
{
    std::lock_guard<std::mutex> lock(mtx_);
    if(data_queue_.empty())
        throw std::runtime_error("queue is empty");
    return data_queue_.front();
}

template<typename T>
const T& ThreadSafeQueue<T>::front() const
{
    std::lock_guard<std::mutex> lock(mtx_);
    if(data_queue_.empty())
        throw std::runtime_error("queue is empty");
    return data_queue_.front();
}

template<typename T>
std::shared_ptr<T> ThreadSafeQueue<T>::front_ptr()
{
    std::lock_guard<std::mutex> lock(mtx_);
    if(data_queue_.empty())
        return std::shared_ptr<T>();
    return std::make_shared<T>(data_queue_.front());
}

#endif // THREADSAFEQUEUE_H