用C++实现数据总线的方法系列(中):数据总线的实现

视频教程:https://www.bilibili.com/video/av95993899/

用C++实现数据总线的方法,本文主要介绍实现一个完整功能的带超时的数据总线DataQueue的方法。

数据总线以及传输的数据的定义和实现

传输数据类Data,它是由数据的地址和数据的大小两个成员组成的。

数据总线类DataQueue,需要有存放总线数据的数据链表,构成锁的互斥量和用于多线程同步的条件变量,同时也需要具有最基本的Push和Pop函数,还有就是Clear和Empty函数。

#ifndef DATA_QUEUE_H
#define DATA_QUEUE_H

#include<stdint.h>
#include<mutex>
#include<condition_variable>
#include<list>

class Data
{
public:
    Data(): m_address(NULL), m_size(0)
    {


    }

    Data(uint64_t size): m_size(size)
    {
        m_address = new char[m_size];
    }

    ~Data()
    {
        if (m_address != NULL)
        {
            delete[] m_address;
            m_address = NULL;
        }
    }

    char* m_address;
    uint64_t m_size;
};
class DataQueue
{
public:
    static void Clear();
    static bool Empty();

    static void Push(Data* dataPtr);
    static void Push(void* data, uint64_t size);
    static Data* Pop();
private:
    static std::mutex               s_mt;
    static std::condition_variable  s_cv;
    static std::list<Data *>        s_queue;
};


#endif
#include "DataQueue.h"

static int POP_TIMEOUT = 1000;

std::mutex              DataQueue::s_mt;
std::condition_variable DataQueue::s_cv;
std::list<Data *>       DataQueue::s_queue;
void DataQueue::Clear()
{
    std::list<Data *>::iterator it = s_queue.begin();
    for (; it != s_queue.end(); ++it)
    {
        delete *it;
    }
    s_queue.clear();
}
bool DataQueue::Empty()
{
    std::unique_lock<std::mutex> lock(s_mt);
    return s_queue.empty();
}

void DataQueue::Push(Data* dataPtr)
{
    std::unique_lock<std::mutex> lock(s_mt);
    s_queue.push_back(dataPtr);
    s_cv.notify_all();
}
void DataQueue::Push(void* data, uint64_t size)
{
    Data *dataPtr = new Data(size);
    memcpy(dataPtr->m_address, data, size);
    Push(dataPtr);
}

Data* DataQueue::Pop()
{
    Data* dataPtr = NULL;
    {
        std::unique_lock<std::mutex> lock(s_mt);
        while (s_queue.empty())
        {
            if (s_cv.wait_for(lock, std::chrono::milliseconds(POP_TIMEOUT)) == std::cv_status::timeout)
            {
                break;
            }
        }

        if (!s_queue.empty())
        {
            dataPtr = s_queue.front();
            s_queue.pop_front();
        }
    }
    
    return dataPtr;

}

这个示例所实现的数据总线呢,虽然可以传递任意类型的数据,但是为了提高效率,所以我们传递的实际上是原始数据的地址,这样减少了数据的构造和拷贝,确实提高了效率,但在使用的过程中就需要留意使用方法,以免造成程序异常。

读取数据会出错的情况

传输的数据类User,它是由id和info两个数据局成员来组成的。同时我们定义了UserPtr作为User类的shared_ptr智能指针。

#ifndef USER_H
#define USER_H

#include<string>
#include<memory>
class User
{
public:
    User(int id, std::string info):m_id(id), m_info(info) {}

    int ID()
    {
        return m_id;
    }

    std::string Info()
    {
        return m_info;
    }
private:
    int m_id;
    std::string m_info;
};

using UserPtr = std::shared_ptr<User>;

调用程序,在SetData线程函数中,我们新建了一个原始数据User对象,并用一个shared_ptr智能指针pUser指向它,然后我们用了一个变量addr存放了这个User对象的地址,然后将addr里的值,也就是User对象的地址放入拷贝到传输数据对象data中,存放到数据总线中。
在GetData线程函数中,获得了传输的数据对象data,然后得到了data里面的值,也就是原始数据User对象的地址,然后使用reinterpret_cast将这个地址值强转为一个目标数据User对象,因为原始对象和目标对象的结构一致,所以也就得到了原始数据值。

#include "stdafx.h"
#include "User.h"
#include "DataQueue.h"
#include<atomic>
#include<thread>
#include<string>
#include<iostream>
static std::atomic_bool getStop = false;

void SetData()
{
    std::this_thread::sleep_for(std::chrono::seconds(1));
    int i = 1;
    while (i <= 1) //insert one data
    {   
        UserPtr pUser(new User(i, std::to_string(i * 10)));
        User* user = pUser.get();
        uint64_t addr = (uint64_t)user;
        DataQueue::Push(&addr, sizeof(addr));
        i++;
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
}


void GetData()
{
    while (true)
    {
        Data *data = DataQueue::Pop();
        if (data != NULL)
        {
            uint64_t addr = 0;
            memcpy(&addr, data->m_address, sizeof(addr));
            UserPtr pUser(reinterpret_cast<User*>(addr));
            User* user = pUser.get();
            std::cout << "---user id: " << user->ID() << "---user info: " << user->Info().c_str() << std::endl;


            //std::cout << "---user id: " << pUser->ID() << "---user info: " << pUser->Info().c_str() << std::endl;
        }


    }
}
int main()
{
    std::thread thGet(GetData);
    std::thread thSet(SetData);


    thGet.join();
    thSet.join();
    return 0;
}

但是报错了,而且是个内存错误,原因是为了提高效率,我们在数据总线上传输的是原始数据的地址,但是,因为在SetData新建的原始数据User使用的是shared_ptr智能指针来指向它,在Push函数放入原始数据地址后,shared_ptr智能指针这个局部变量也将释放,这样原始数据的引用计数就归零了,所以原始数据就被释放了,导致读取线程在总线Pop数据时崩溃。

正确的数据总线的使用方法

正确的使用方法是:使用boost::intrusive_ptr智能指针,同时需要用总线的数据类自己来实现引用计数,这样就可以保证原始数据不被释放,具体方法如下:

建立一个提供引用计数功能的基类ReferenceCounter,这个基类需要提供intrusive_ptr_add_ref和intrusive_ptr_release两个接口供boost::intrusive_ptr来进行引用计数的增加和减少。同时,要提供两个接口来供我们自己来进行进行引用计数的增加和减少,这里我们实现了IncReference和DecReference这两个接口来完成相应的功能。

#ifndef ADVANCED_USER_H
#define ADVANCED_USER_H
#include <assert.h>
#include <atomic>
#include <string>
#include <boost/intrusive_ptr.hpp>

class ReferenceCounter
{
public:
    friend void intrusive_ptr_add_ref(ReferenceCounter *p)
    {
        assert(p);
        assert(p->ref_count >= 0);
        ++p->ref_count;
    }

    friend void intrusive_ptr_release(ReferenceCounter *p)
    {
        assert(p);
        assert(p->ref_count > 0);
        if (--p->ref_count == 0)
        {
            delete p;
        }
    }
    friend uint64_t IncReference(ReferenceCounter *p)
    {
        intrusive_ptr_add_ref(p);
        return reinterpret_cast<uint64_t>(p);
    }

    friend void DecReference(ReferenceCounter *p)
    {
        intrusive_ptr_release(p);
    }

    int RefCount()
    {
        return ref_count;
    }
protected:
    ReferenceCounter() { ref_count = 0; }
    virtual ~ReferenceCounter() {};

private:
    std::atomic_int ref_count;
};

class AdvancedUser: public ReferenceCounter
{
public:
    AdvancedUser(int id, std::string info) :m_id(id), m_info(info) {}

    int ID()
    {
        return m_id;
    }

    std::string Info()
    {
        return m_info;
    }
private:
    int m_id;
    std::string m_info;
};

using AdvancedUserPtr = boost::intrusive_ptr<AdvancedUser>;
#endif

如果调用总线的方式不改变,还是会报错,因为其实和之前使用shared_ptr没有本质的区别,还是会因为数据释放问题而报错,只不过我们能更清楚的了解数据释放的过程。

#include "stdafx.h"
#include "User.h"
#include "AdvancedUser.h"
#include "DataQueue.h"
#include<atomic>
#include<thread>
#include<string>
#include<iostream>

void SetData()
{
    std::this_thread::sleep_for(std::chrono::seconds(1));
    int i = 1;
    while (i <= 1) 
    {
        AdvancedUserPtr pUser(new AdvancedUser(i, std::to_string(i * 10)));
        AdvancedUser*  user = pUser.get();
        uint64_t addr = (uint64_t)user;
        DataQueue::Push(&addr, sizeof(addr));
        i++;
    }
    std::cout << "---SetData End!" << std::endl;
}

void GetData()
{
    while (true)
    {
        std::this_thread::sleep_for(std::chrono::seconds(5));
        Data *data = DataQueue::Pop();
        if (data != NULL)
        {
            uint64_t addr = 0;
            memcpy(&addr, data->m_address, sizeof(data->m_size));
            AdvancedUserPtr pUser(reinterpret_cast<AdvancedUser*>(addr));
            std::cout << "---user id: " << pUser->ID() << "---user info: " << pUser->Info().c_str() << std::endl;
        }
    }
    std::cout << "---GetData End!" << std::endl;
}


int main()
{
    std::thread thGet(GetData);
    std::thread thSet(SetData);

    thGet.join();
    thSet.join();
    return 0;
}
  • 在建立智能指针对象时,对象的引用计数加1后为1

在push完成,智能指针对象出作用域之后,对象的引用计数减1后为0,释放了原始对象,在Pop数据时同样会崩溃

正确的使用方法如下,在SetData线程函数里的Push数据到总线之前,调用IncReference,使得对象的计数加1,这样原始数据的引用计数就为2了,当智能指针出作用域,对象计数减1后,原始数据的引用计数为从2变为1,但没有清零,所以原始数据比不会被释放。

当然,那在GetData函数中的得到原始数据之后,还需要调用DecReference将对象计数减1,也就是减掉我们之前人为调用IncReference的那个计数加1。代码如下:

#include "stdafx.h"
#include "User.h"
#include "AdvancedUser.h"
#include "DataQueue.h"
#include<atomic>
#include<thread>
#include<string>
#include<iostream>

static bool StopFlag = false;
void SetData()
{
    std::this_thread::sleep_for(std::chrono::seconds(1));
    int i = 1;
    while (i <= 10) 
    {
        AdvancedUserPtr pUser(new AdvancedUser(i, std::to_string(i * 10)));
        AdvancedUser*  user = pUser.get();
        uint64_t addr = (uint64_t)user;
        IncReference(pUser.get());
        std::cout << "Set---user id: " << pUser->ID() << "---user info: " << pUser->Info().c_str() << std::endl;
        DataQueue::Push(&addr, sizeof(addr));
        i++;
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
    std::cout << "---SetData End!" << std::endl;
    getStop = true;
}
void GetData()
{
    while (!getStop)
    {
        //std::this_thread::sleep_for(std::chrono::seconds(1));
        Data *data = DataQueue::Pop();
        if (data != NULL)
        {
            uint64_t addr = 0;
            memcpy(&addr, data->m_address, sizeof(data->m_size));
            AdvancedUserPtr pUser(reinterpret_cast<AdvancedUser*>(addr));
            std::cout << "Get---user id: " << pUser->ID() << "---user info: " << pUser->Info().c_str() << std::endl;
            DecReference(pUser.get());
        }
    }
    std::cout << "---GetData End!" << std::endl;
}

int main()
{
    std::thread thGet(GetData);
    std::thread thSet(SetData);


    thGet.join();
    thSet.join();
    return 0;
}

//output
Set---user id: 1---user info: 10
Get---user id: 1---user info: 10
Set---user id: 2---user info: 20
Get---user id: 2---user info: 20
Set---user id: 3---user info: 30
Get---user id: 3---user info: 30
Set---user id: 4---user info: 40
Get---user id: 4---user info: 40
Set---user id: 5---user info: 50
Get---user id: 5---user info: 50
Set---user id: 6---user info: 60
Get---user id: 6---user info: 60
Set---user id: 7---user info: 70
Get---user id: 7---user info: 70
Set---user id: 8---user info: 80
Get---user id: 8---user info: 80
Set---user id: 9---user info: 90
Get---user id: 9---user info: 90
Set---user id: 10---user info: 100
Get---user id: 10---user info: 100
---SetData End!
---GetData End!
最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 201,784评论 5 474
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 84,745评论 2 378
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 148,702评论 0 335
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,229评论 1 272
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,245评论 5 363
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,376评论 1 281
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 37,798评论 3 393
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,471评论 0 256
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 40,655评论 1 295
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,485评论 2 318
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,535评论 1 329
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,235评论 3 318
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 38,793评论 3 304
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,863评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,096评论 1 258
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 42,654评论 2 348
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,233评论 2 341

推荐阅读更多精彩内容