快捷導(dǎo)航

C++線程安全容器stack和queue的使用詳細介紹

更新時間：2022年08月23日 08:56:51 作者：TTOR

stack是一種容器適配器，專門用在具有后進先出操作的上下文環(huán)境中，其刪除只能從容器的一端進行元素的插入與提取操作；隊列是一種容器適配器，專門用于在FIFO上下文(先進先出)中操作，其中從容器一端插入元素，另一端提取元素

線程安全的容器棧threadsafe_stack

入門(3)里曾介紹過線程安全的stack容器, 這里把代碼搬過來再分析

逐項分析, 該代碼如何實現(xiàn)線程安全的

template<typename T>
class threadsafe_stack
{
private:
    stack<T> data;
    mutable mutex m;
public:
    threadsafe_stack(){}
    threadsafe_stack(const threadsafe_stack &other)
    {
        lock_guard lock1(other.m);
        data=other.data;
    }
    threadsafe_stack &operator=(const threadsafe_stack &) = delete;
    void push(T new_value)
    {
        lock_guard lock1(m);
        data.push(move(new_value));      //1
    }
    shared_ptr<T> pop()
    {
        lock_guard lock1(m);
        if (data.empty())
        {
            throw empty_stack();        //2
        }
        shared_ptr<T> const
                res(make_shared<T>(move(data.top()))); //3
        data.pop();                                     //4
        return res;
    }
    void pop(T &value)
    {
        lock_guard lock1(m);
        if (data.empty())
        {
            throw empty_stack();
        }
        value = move(data.top());    //5
        data.pop();                 //6
    }
    bool empty() const //7
    {
        lock_guard lock1(m);
        return data.empty();
    }
};

首先, 每個操作都對互斥加鎖, 保證基本線程安全

其次, 在多線程下, 對于std::stack容器, empty(), top(), pop()存在接口上的數(shù)據(jù)競爭(見入門(3)說明), 于是threadsafe_stack把這些調(diào)用集合到一個函數(shù)pop()里, 以實現(xiàn)線程安全. 其中pop()函數(shù)里若與遇?？? 直接拋出異常

接著分析:

1處data.push()可能拋出異常: 原因是復(fù)制/移動時拋出異?；騭tack容器擴展容量時遇上內(nèi)存分配不足, 但無論哪種, std::stack<>能保證自身的安全

2處拋出的異常: 沒有改動數(shù)據(jù), 安全的拋出行為

3處共享指針的創(chuàng)建可能拋出異常: 內(nèi)存不足或移動/復(fù)制相關(guān)的構(gòu)造函數(shù)拋出異常,但兩種情形c++都能保證不會出現(xiàn)內(nèi)存泄漏, 并且此時數(shù)據(jù)還未改動(data.pop()時才改動),

4處data.pop()的實質(zhì)操作是返回結(jié)果, 絕不會拋出異常,結(jié)合3, 所以這是異常安全的重載函數(shù)pop()

5,6處和3,4處類似, 不同之處是沒用創(chuàng)建新共享指針, 但此時數(shù)據(jù)也沒被改動, 也是安全的重載函數(shù)pop()

最后7處empty()不改動任何數(shù)據(jù), 是異常安全的函數(shù)

從內(nèi)存和數(shù)據(jù)結(jié)構(gòu)安全方面來說沒用問題

然而,這段代碼可能造成死鎖:

因為在持鎖期間, 有可能執(zhí)行以下用戶自定義函數(shù):

用戶自定義的復(fù)制構(gòu)造函數(shù)(1 3處的res構(gòu)造), 移動構(gòu)造函數(shù)(3處的make_share), 拷貝賦值操作和移動賦值操作(5處), 用戶也可能重載了new和delete.

當(dāng)在這些函數(shù)里, 若是再次調(diào)用了同個棧的相關(guān)函數(shù), 會再次申請獲取鎖, 然而之前的鎖還沒釋放, 因此造成死鎖

以下是我想到的一種死鎖方式(正常情況應(yīng)該不會這么寫, 但是設(shè)計時必須要考慮)

class A;
threadsafe_stack<A> s;
class A
{
public:
    A(A&& a)//2->然后這里使用s.pop(),之前鎖沒釋放, 造成了死鎖
    {
        s.pop();
    }
    A(){}
};
int main()
{
    s.push(A()); //1->臨時對象A()在s.push()里被move進內(nèi)置data時, 會調(diào)用A的移動構(gòu)造函數(shù)
    return 0;
}

向棧添加/移除數(shù)據(jù), 不可能不涉及復(fù)制行為或內(nèi)存行為, 于是只能對棧的使用者提出要求: 讓使用者來保證避免死鎖

棧的各成員函數(shù)都有l(wèi)ock_guard保護數(shù)據(jù), 因此同時調(diào)用的線程沒有數(shù)量限制.

僅有構(gòu)造函數(shù)和析構(gòu)函數(shù)不是安全行為, 但無論是沒構(gòu)造完成還是銷毀到一半, 從而轉(zhuǎn)去調(diào)用成員函數(shù), 這在有無并發(fā)情況下都是不正確的.

所以, 使用者必須保證: 棧容器未構(gòu)造完成時不能訪問數(shù)據(jù), 只有全部線程都停止訪問時, 才可銷毀容器

線程安全的容器隊列threadsafe_queue

自定義一個threadsafe_queue, 并且上面對于線程安全的大多數(shù)分析在這也成立

template<typename T>
class threadsafe_queue
{
private:
    queue<T> data;
    mutable mutex m;
    condition_variable condition;
public:
    threadsafe_queue()
    {}
    threadsafe_queue(const threadsafe_queue &other)
    {
        lock_guard lock1(other.m);
        data = other.data;
    }
    threadsafe_queue &operator=(const threadsafe_queue &) = delete;
    void push(T new_value)
    {
        lock_guard lock1(m);
        data.push(move(new_value));
        condition.notify_one();   //1
    }
    void wait_and_pop(T &value)      //2
    {
        lock_guard lock1(m);
        condition.wait(lock1, [this]
        {
            return !data.empty();
        });
        value = move(data.top());
        data.pop();
    }
    shared_ptr<T> wait_and_pop()       //3
    {
        lock_guard lock1(m);
        condition.wait(lock1, [this]
        {
            return !data.empty();
        });
        shared_ptr<T> const
                res(make_shared<T>(move(data.top()))); //4 創(chuàng)建shared_ptr可能出現(xiàn)異常
        data.pop();
        return res;
    }
    shared_ptr<T> try_pop()
    {
        lock_guard lock1(m);
        if (data.empty())
        {
            return shared_ptr<T>();     //5
        }
        shared_ptr<T> const
                res(make_shared<T>(move(data.top())));
        data.pop();
        return res;
    }
    bool try_pop(T &value)
    {
        lock_guard lock1(m);
        if (data.empty())
        {
            return false;
        }
        value = move(data.top());
        data.pop();
    }
    bool empty() const
    {
        lock_guard lock1(m);
        return data.empty();
    }
};

區(qū)別:

發(fā)現(xiàn)隊列通常用于消費者/生產(chǎn)者模型, 因此實現(xiàn)阻塞的取值函數(shù)wait_and_pop, 即當(dāng)調(diào)用時隊列若空, 阻塞等待, 直到push數(shù)據(jù)后調(diào)用condition.notify_one()

同時也提供了非阻塞的取值函數(shù)try_pop

然而這一實現(xiàn)會有問題:

假如有多個線程同時等待, condition.notify_one()只能喚醒其中一個,若該喚醒的線程執(zhí)行wait_and_pop之后的代碼拋出異常(例如4處res的創(chuàng)建), 此時隊列里還有數(shù)據(jù),卻不會有其他任何線程被喚

如果我們因不能接受這種行為方式, 而只是簡單的把notify_one改為notify_all,這樣每次push數(shù)據(jù)后都會喚醒所有的等待線程. 由于只push了1個數(shù)據(jù), 大多數(shù)線程醒來后發(fā)現(xiàn)隊列還是為空, 還得繼續(xù)等待, 這將大大增加開銷

第二種解決種方法是若wait_and_pop拋出異常則再次調(diào)用notify_one

第三種方法是讓std::queue存儲share_ptr<T>, share_ptr的初始化移動到push的調(diào)用處, 從內(nèi)部復(fù)制shared_ptr<>實例則不會拋出異常

這里采用第三種方法, 還會有額外的好處: push里為shared_ptr分配內(nèi)存操作在加鎖之前, 縮短了互斥加鎖的時間, 由于分配內(nèi)存通常是耗時的操作, 因此這樣非常有利于增強性能

template<typename T>
class threadsafe_queue
{
private:
    queue<shared_ptr<T>> data;
    mutable mutex m;
    condition_variable condition;
public:
    threadsafe_queue()
    {}
    threadsafe_queue(const threadsafe_queue &other)
    {
        lock_guard lock1(other.m);
        data = other.data;
    }
    threadsafe_queue &operator=(const threadsafe_queue &) = delete;
    void push(T new_value)
    {
        //分配內(nèi)存在加鎖操作之前
        shared_ptr<T> value(make_shared<T>(move(new_value)));
        lock_guard lock1(m);
        data.push(value);
        condition.notify_one();
    }
    void wait_and_pop(T &value)
    {
        lock_guard lock1(m);
        condition.wait(lock1, [this]
        {
            return !data.empty();   //隊列空則等待
        });
        value = move(*data.front()); //先取值, 再存入?yún)?shù)value
        data.pop();
    }
    bool try_pop(T &value)
    {
        lock_guard lock1(m);
        if (data.empty())
        {
            return false;       //隊列空返回false
        }
        value = move(*data.front()); //先取值, 再存入?yún)?shù)value
        data.pop();
        return true;
    }
    shared_ptr<T> wait_and_pop()
    {
        lock_guard lock1(m);
        condition.wait(lock1, [this]
        {
            return !data.empty();    //隊列空則等待
        });
        shared_ptr<T> res = data.front(); //取出結(jié)果返回給外部
        data.pop();
        return res;
    }
    shared_ptr<T> try_pop()
    {
        lock_guard lock1(m);
        if (data.empty())
        {
            return shared_ptr<T>();     //隊列空返回空shared_ptr
        }
        shared_ptr<T> res = data.front();//取出結(jié)果返回給外部
        data.pop();
        return res;
    }
    bool empty() const
    {
        lock_guard lock1(m);
        return data.empty();
    }
};

到此這篇關(guān)于C++線程安全容器stack和queue的使用詳細介紹的文章就介紹到這了,更多相關(guān)C++ stack和queue內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: