快捷導(dǎo)航

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

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

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

線程安全的容器棧threadsafe_stack

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

逐項(xiàng)分析, 該代碼如何實(shí)現(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();
    }
};

首先, 每個(gè)操作都對(duì)互斥加鎖, 保證基本線程安全

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

接著分析:

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

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

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

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

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

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

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

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

因?yàn)樵诔宙i期間, 有可能執(zhí)行以下用戶自定義函數(shù):

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

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

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

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

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

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

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

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

線程安全的容器隊(duì)列threadsafe_queue

自定義一個(gè)threadsafe_queue, 并且上面對(duì)于線程安全的大多數(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)隊(duì)列通常用于消費(fèi)者/生產(chǎn)者模型, 因此實(shí)現(xiàn)阻塞的取值函數(shù)wait_and_pop, 即當(dāng)調(diào)用時(shí)隊(duì)列若空, 阻塞等待, 直到push數(shù)據(jù)后調(diào)用condition.notify_one()

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

然而這一實(shí)現(xiàn)會(huì)有問(wèn)題:

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

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

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

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

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

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();   //隊(duì)列空則等待
        });
        value = move(*data.front()); //先取值, 再存入?yún)?shù)value
        data.pop();
    }
    bool try_pop(T &value)
    {
        lock_guard lock1(m);
        if (data.empty())
        {
            return false;       //隊(duì)列空返回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();    //隊(duì)列空則等待
        });
        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>();     //隊(duì)列空返回空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的使用詳細(xì)介紹的文章就介紹到這了,更多相關(guān)C++ stack和queue內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: