C++并发编程 等待与唤醒 条件变量
条件变量, 包括(std::condition_variable 和 std::condition_variable_any)
定义在 condition_variable 头文件中, 它们都需要与互斥量(作为同步工具)一起才能工作.
std::condition_variable 允许阻塞一个线程, 直到条件达成.
成员函数
void wait(std::unique_lock<std::mutex>& lock);
等待, 通过 notify_one(), notify_all()或伪唤醒结束等待
void wait(std::unique_lock<std::mutex>& lock, Predicate pred);
等待, 通过 notify_one(), notify_all()被调用, 并且谓词为 true 时结束等待.
pred 谓词必须是合法的, 并且需要返回一个值, 这个值可以和bool互相转化
cv_status wait_until(std::unique_lock<std::mutex>& lock, const std::chrono::time_point<Clock, Duration>& absolute_time);
调用 notify_one(), notify_all(), 超时或线程伪唤醒时, 结束等待.
返回值标识了是否超时.
bool wait_until(std::unique_lock<std::mutex>& lock, const std::chrono::time_point<Clock, Duration>& absolute_time, Predicate pred);
等待, 通过 notify_one(), notify_all(), 超时, 线程伪唤醒, 并且谓词为 true 时结束等待.
cv_status wait_for(std::unique_lock<std::mutex>& lock, const std::chrono::duration<Rep, Period>& relative_time);
调用 notify_one(), notify_all(), 指定时间内达成条件或线程伪唤醒时，结束等待
bool wait_for(std::unique_lock<std::mutex>& lock, const std::chrono::duration<Rep, Period>& relative_time, Predicate pred);
调用 notify_one(), notify_all(), 指定时间内达成条件或线程伪唤醒时，并且谓词为 true 时结束等待.
void notify_one() noexcept; 唤醒一个等待当前 std::condition_variable 实例的线程
void notify_all() noexcept; 唤醒所有等待当前 std::condition_variable 实例的线程
一个线程安全的队列设计:
#ifndef _THREAD_SAFE_QUEUE_ #define _THREAD_SAFE_QUEUE_ #include <condition_variable> #include <mutex> #include <queue> #include <memory> template<typename Ty, typename ConditionVar = std::condition_variable, typename Mutex = std::mutex> class ThreadSafeQueue { typedef std::queue<Ty> Queue; typedef std::shared_ptr<Ty> SharedPtr; typedef std::lock_guard<Mutex> MutexLockGuard; typedef std::unique_lock<Mutex> MutexUniqueLock; public: explicit ThreadSafeQueue() {} ~ThreadSafeQueue() {} ThreadSafeQueue(const ThreadSafeQueue&) = delete; ThreadSafeQueue& operator=(const ThreadSafeQueue&) = delete; bool IsEmpty() const { MutexLockGuard lk(mMutex); return mQueue.empty(); } void WaitAndPop(Ty& value) { MutexUniqueLock lk(mMutex); mConVar.wait(lk, [this] { return !mQueue.empty(); }); value = mQueue.front(); mQueue.pop(); } SharedPtr WaitAndPop() { MutexUniqueLock lk(mMutex); mConVar.wait(lk, [this] { return !mQueue.empty(); }); SharedPtr sp = std::make_shared<Ty>(mQueue.front()); mQueue.pop(); return sp; } bool TryPop(Ty& value) { MutexLockGuard lk(mMutex); if (mQueue.empty()) return false; value = mQueue.front(); mQueue.pop(); return true; } SharedPtr TryPop() { MutexLockGuard lk(mMutex); if (mQueue.empty()) return false; SharedPtr sp = std::make_shared<Ty>(mQueue.front()); mQueue.pop(); return sp; } void Push(const Ty& value) { MutexLockGuard lk(mMutex); mQueue.push(value); mConVar.notify_all(); } private: mutable Mutex mMutex; ConditionVar mConVar; Queue mQueue; }; #endif // _THREAD_SAFE_QUEUE_
另一个版本, 使用 shared_ptr 作为成员对队列的性能有很大的提升, 其在push时减少了互斥量持有的时间, 允许其它线程在分配内存的同时,对队列进行其它操作.
template<typename Ty, typename ConditionVar = std::condition_variable, typename Mutex = std::mutex> class ThreadSafeQueue { typedef std::shared_ptr<Ty> SharedPtr; typedef std::queue<SharedPtr> Queue; typedef std::shared_ptr<Ty> SharedPtr; typedef std::lock_guard<Mutex> MutexLockGuard; typedef std::unique_lock<Mutex> MutexUniqueLock; public: explicit ThreadSafeQueue() {} ~ThreadSafeQueue() {} ThreadSafeQueue(const ThreadSafeQueue&) = delete; ThreadSafeQueue& operator=(const ThreadSafeQueue&) = delete; bool IsEmpty() const { MutexLockGuard lk(mMutex); return mQueue.empty(); } void WaitAndPop(Ty& value) { MutexUniqueLock lk(mMutex); mConVar.wait(lk, [this] { return !mQueue.empty(); }); value = std::move(*mQueue.front()); mQueue.pop(); } SharedPtr WaitAndPop() { MutexUniqueLock lk(mMutex); mConVar.wait(lk, [this] { return !mQueue.empty(); }); SharedPtr sp = mQueue.front(); mQueue.pop(); return sp; } bool TryPop(Ty& value) { MutexLockGuard lk(mMutex); if (mQueue.empty()) return false; value = std::move(*mQueue.front()); mQueue.pop(); return true; } SharedPtr TryPop() { MutexLockGuard lk(mMutex); if (mQueue.empty()) return false; SharedPtr sp = mQueue.front(); mQueue.pop(); return sp; } void Push(const Ty& value) { SharedPtr p = std::make_shared<Ty>(value); MutexLockGuard lk(mMutex); mQueue.push(p); mConVar.notify_all(); } private: mutable Mutex mMutex; ConditionVar mConVar; Queue mQueue; };
std::future