时间:2022-08-04 10:03:50 | 栏目:C代码 | 点击:次
#include<queue> #include<mutex> #include<condition_variable> #include<optional> #include<cassert> #include<thread> template<typename T,typename Container = std::queue<T>> class Queue //无界队列 { public: Queue() = default; ~Queue() = default; //禁止拷贝和移动,编译器会自动delete /*Queue(const Queue&) = delete; Queue(Queue&&) = delete; Queue& operator=(const Queue&) = delete; Queue& operator=(Queue&&) = delete;*/ void push(const T& val) { emplace(val); } void push(T&& val) { emplace(std::move(val)); } template<typename...Args> void emplace(Args&&...args) { std::lock_guard lk{ mtx_ }; q_.push(std::forward<Args>(args)...); cv_.notify_one(); } T pop()//阻塞 { std::unique_lock lk{ mtx_ }; cv_.wait(lk, [this] {return !q_.empty(); });//如果队列不为空就继续执行,否则阻塞 assert(!q_.empty()); T ret{ std::move_if_noexcept(q_.front()) }; q_.pop(); return ret; } std::optional<T> try_pop()//非阻塞 { std::unique_lock lk{ mtx_ }; if (q_.empty())return {}; std::optional<T> ret{ std::move_if_noexcept(q_.front()) }; q_.pop(); return ret; } bool empty()const { std::lock_guard lk{ mtx_ }; return q_.empty(); } private: Container q_; mutable std::mutex mtx_; std::condition_variable cv_; }; #include<iostream> int main() { Queue<int>q; std::thread t1( [&] { for (int i = 0; i < 100; ++i) { q.push(i); } }); std::thread t2( [&] { for (int i = 0; i < 100; ++i) { //std::cout<<q.pop()<<" "; if (auto ret = q.try_pop()) { std::cout << *ret<<" "; } } }); t1.join(); t2.join(); return 0; }
#include<mutex> #include<condition_variable> #include<boost/circular_buffer.hpp> #include<optional> template<typename T> class Queue { public: Queue(size_t capacity) :q_{ capacity }{} template<typename T> void push(T&& val)//阻塞 { std::unique_lock lk{ mtx_ }; not_full_.wait(lk, [this] {return !q_.full(); }); assert(!q_.full()); q_.push_back(std::move(std::forward<T>(val))); not_empty_.notify_one(); } template<typename T> bool try_push(T&& val)//非阻塞 { std::lock_guard lk{ mtx_ }; if (q_.full())return false; q_.push_back(std::forward<T>(val)); not_empty_.notify_one(); return true; } T pop()//阻塞 { std::unique_lock lk{ mtx_ }; not_empty_.wait(lk, [this] {return !q_.empty(); }); asert(!q_.empty()); T ret{ std::move_if_noexcept(q_.front()) }; q_.pop_front(); not_full_.notify_one(); return ret; } std::optional<T> try_pop()//非阻塞 { std::lock_guard lk{ mtx_ }; if (q_.empty())return {}; std::optional<T> ret{ std::move_if_noexcept(q_.front()) }; q_.pop_front(); not_full_.notify_one(); return ret; } private: boost::circular_buffer<T>q_; std::mutex mtx_; std::condition_variable not_full_; std::condition_variable not_empty_; }; #include<iostream> int main() { Queue<int>q(10); std::thread t1( [&] { for (int i = 0; i < 100; ++i) { q.push(i); } }); std::thread t2( [&] { for (int i = 0; i < 100; ++i) { //std::cout<<q.pop()<<" "; if (auto ret = q.try_pop()) { std::cout << *ret << " "; } } }); t1.join(); t2.join(); return 0; }