当前位置：首页 > news >正文

4.采用锁操作并支持等待功能的线程安全队列

news 来源：原创 2025/5/13 21:50:14

分析

书接上文

修改push()似乎并不困难:在函数末尾加上对data_cond.notify_one()的调用即可，与代码清单1(第一篇文章)一样。事情其实没那么简单，我们之所以采用精细粒度的锁，目的是尽可能提高并发操作的数量。如果在notify_one()调用期间，互斥依然被锁住，形式与代码清单1一样，而等待通知的线程却在互斥解锁前觉醒，它就需要继续等待互斥解锁。矛盾的是，若在解锁互斥之后调用notify_one()，那么互斥已经可以再次获取，并且超前一步，等着接受通知的线程对其加锁(前提是其他线程没有抢先将其重新锁住)。这点改进看似细微，但对某些情况却有重要作用。

wait_and_pop()就复杂得多，因为我们需要确定在哪里等待、根据什么断言唤醒等待、需要锁住什么互斥等。等待唤醒的条件是“队列非空”，用代码表示为 head!=tail。
按这种写法，要求两个互斥head_mutex和 tail_mutex 都被锁住，我们分析代码清单3(第三篇文章)的时候就已经确定,只有在读取 tail 指针时才有必要锁住互斥tai_mutex,而比较运算无须保护，本例同理。若我们将断言设定为head!=get_tail(),则只需持有互斥 head_mutex，所以在调用 data_cond.wait()时，就可以重新锁住head_mutex。只要我们加入了等待的逻辑，这种实现就与 try_pop()一样。
对于try_pop()的另一个重载和对应的 wait_and_pop() 的重载，我们也要谨慎思考和设计。在代码清单3中，try_pop()函数的结果通过共享指针std::shared_ptr的实例返回，其指向目标由old_head间接从 pop_head() 取得。如果模仿代码清单1，将以上方法改为 try_pop()的第一个重载的模式，令函数接收名为value 的引用参数，再由拷贝赋值操作赋予它 old_head 的值，就可能会出现与异常有关的问题。根据这种改动，在拷贝赋值操作执行时，数据已经从队列中移除，且互斥已经解锁，剩下的全部动作就是将数据返回给调用者。但是，如果拷贝赋值操作抛出了异常(完全有可能)，则该项数据丢失，因为它无法回到队列本来的位置上。
若队列模板在具现化时，模板参数采用了实际类型T，而该类型支持不抛出异常的移动赋值操作，或不抛出异常的交换操作，我们即可使用类型T。然而，我们还是更希望实现通用的解决方法，对任何类型T都有效。在上述场景中，我们需要在队列移除节点以前，将可能抛出异常的操作移动到锁的保护范围之内。也就是说，我们还需要pop_head()的另一个作用，在改动队列之前就获取其存储的值。
相比而言，empty(就很简单了:只需锁住互斥head_mutex，然后检查 head_get_tail()。

一、实现

接口

template<typename T>
class threadsafe_queue{
private:struct node{std::shared_ptr<T> data;std::unique_ptr<node> next;};std::mutex head_mutex;std::unique_ptr<node> head;std::mutex tail_mutex;node* tail;std::condition_variable data_cond;
public:threadsafe_queue():head(new node),tail(head.get()){}threadsafe_queue(const threadsafe_queue& other)=delete;threadsafe_queue& operator=(const threadsafe_queue& other)=delete;std::shared_ptr<T> try_pop();bool try_pop(T& value);std::shared_ptr<T>wait_and_pop();void wait_and_pop(T& value);void push(T new_value);bool empty();
];

压入新数据

template<typename T>{void threadsafe_queue<T>::push(T new_value)std::shared_ptr<T> new_data(std::make_shared<T>(std::move(new_value)));std::unique_ptr<node>p(new node);{std::lock_guard<std::mutex> tail_lock(tail_mutex);tail->data=new_data;node* const new_tail=p.get();tail->next=std::move(p);tail=new_tail;}data_cond.notify_one();
}

我们提过，复杂之处全在于pop()上，我们运用几个辅助函数简化操作。

template<typename T>
class threadsafe_queue{
private :node* get_tail(){std::lock_guard<std::mutex> tail_lock(tail mutex);return tail;}std::unique_ptr<node> pop_head()①{std::unique_ptr<node> old_head=std::move(head);head=std::move(old_head->next);return old_head;}std::unique_lock<std::mutex> wait_for_data()②{std::unique_lock<std::mutex> head_lock(head_mutex);data_cond.wait(head_lock,[&]{return head.get()!=get_tail();));return std::move(head_lock);③}std::unique_ptr<node> wait_pop_head(){std::unique_lock<std::mutex> head_lock(wait_for_data());④return pop_head();}std::unique ptr<node> wait_pop_head(T& value){std::unique_lock<std::mutex>head_lock(wait_for_data());⑤value=std::move(*head->data);return pop head();}
public:std::shared ptr<T> wait_and_pop(){std::unique_ptr<node> const old_head=wait_pop_head();return old_head->data;}void wait_and_pop(T& value){std::unique_ptr<node> const old_head=wait_pop_head(value);}
};

这里展示出 wait_and_pop()实现代码，它含有几个辅助函数，用以简化代码和减少重复，如 pop_head()①和 wait_for_data()②。前者移除头节点而改动队列,后者则等待数据加入空队列，以将其弹出。wait_for_data()特别值得注意，它在条件变量上等待，以 lambda 函数作为断言，并且向调用者返回锁的实例③。因为 wait pop_head()的两个重载都会改动队列数据，并且都依赖wait_for_data()函数，而后者将锁返回则保证了头节点弹出的全过程都持有同一个锁④⑤。这也为里的 pop_head()和ty_pop()使用。

template<typename T>
class threadsafe_queue
private:std::unique_ptr<node> try_pop_head(){std::lock guard<std::mutex> head lock(head_mutex);if(head.get()==get_tail()){return std::unique_ptr<node>();}return pop_head();}std::unique_ptr<node> try_pop_head(T& value){std::lock guard<std::mutex> head_lock(head_mutex)if(head.get()==get_tail()){return std::unique_ptr<node>();}value=std::move(*head->data);return pop_head()}
public:std::shared ptr<T> try_pop(){std::unique_ptr<node> old_head=try_pop_head();return old_head?old_head->data:std::shared_ptr<T>();}bool try_pop(T& value){std::unique ptr<node> const old_head=try_pop_head(value);return old_head!=nullptr;}bool empty(){std::lock_guard<std::mutex> head_lock(head_mutex);return(head.get()==get_tail());}
);

分析

一、实现

总结

相关文章：