处理大量数据的一种方法(C++)

处理场景：三台线阵相机或者多台其他传感器与工控机连着，通过调用相机的SDK控制相机，例如同时开关，同时从相机中获取数据。然后采集软件接收到相机数据后，将数据保存到硬盘。

应用场景

处理的问题：如果三台相机发送数据频率很快，并且数据量很大的话，如何快速将数据写入到硬盘里或者对数据的其他处理，从而保证内存不增长？

解决方案简述：每收到一坨相机的数据，就将其从尾部压入队列；同时有另外一个线程在等着处理这个队列，只要这个队列有数据，就从头部弹出一个数据，然后对这个数据进行相应的处理。这里存在一个问题：如果相机频率太快，也就是说队列会涨得非常快，涨数据的速度大于处理数据的速度，内存就会上涨，直至崩溃。处理数据，例如将相机数据保存到硬盘，如果硬盘写数据的速度很慢，就会出现内存涨爆的情况，这个时候可以考虑用多线程进行处理，从队列弹出一个数据后，马上被一个线程接上，这样可以显著提升CPU的利用率，加快数据处理速度。

队列缓存

简要代码如下：

需特别注意代码中的注释。

//头文件代码
#include <iostream>
#include <windows.h>
#include <queue>
#include <mutex>
#include <future>
#include "SThreadPool.h"
struct CameraData
{
    int height;
    int width;
    int datasize;
    std::unique_ptr<char[]> data;
    UINT64 timestamps;
};

class Sensor
{
public:
    Sensor();
    ~Sensor();
    //**获取相机数据的回调，并在回调中将相机数据压入容器中**//
    friend int WINAPI s_ManagaGetImageCallBack(void* _context, void* _system, void* _data);

    //从队列头部弹出数据并进行处理//
    void t_ProcessDataQueue(void* para);

    void SaveImage(std::shared_ptr<CameraData> data);
    friend void t_SaveCameraImage(void* para, std::shared_ptr<CameraData> data);
private:

    //用来存储三台相机数据的容器//
    std::mutex m_DataMtx_;
    std::queue<std::shared_ptr<CameraData>> m_QueCamData_;
    std::condition_variable m_DataVar_;
    bool m_bQueExit_ = false;
    std::shared_ptr<SThreadPool> m_Pool_;
    std::future<void> m_fuProcessDataQue_;
};

实现文件中，重点关注函数t_ProcessDataQueue(void* para);同时这段代码还包括一种有效退出类内线程的方式，请看析构函数部分。

//实现文件
Sensor::Sensor()
{
    m_fuProcessDataQue_ = std::async(std::launch::async, &Sensor::t_ProcessDataQueue, this, nullptr);

    unsigned long hardwares = std::thread::hardware_concurrency();
    m_Pool_ = std::make_shared<SThreadPool>(hardwares);
}

Sensor::~Sensor()
{
    {
        std::unique_lock<std::mutex> locker(m_DataMtx_);
        m_bQueExit_ = true;
    }
    m_DataVar_.notify_one();
    m_fuProcessDataQue_.get();
}

int WINAPI s_ManagaGetImageCallBack(void* _context, void* _system, void* _data)
{
    Sensor* pThis = (Sensor*)_context;
    std::shared_ptr<CameraData> data = std::make_shared<CameraData>();

    //将从相机返回出来的数据组装成CameraData.....
    {
        std::unique_lock<std::mutex> lock(pThis->m_DataMtx_);
        pThis->m_QueCamData_.push(data);
    }
    pThis->m_DataVar_.notify_one();//通知另外一个线程进行数据的处理
    return 0;
}

void Sensor::t_ProcessDataQueue(void* para)
{
    std::shared_ptr<CameraData> tempdata;


    std::unique_lock<std::mutex> locker(m_DataMtx_);
    while (true)
    {
        try
        {
            if (!m_bQueExit_) m_DataVar_.wait(locker);
            locker.unlock();
            while (true)
            {
                locker.lock();
                if (m_QueCamData_.empty())
                {
                    std::cout << "camera set queue is empty.\n";
                    break;
                }
                else
                {
                    tempdata = m_QueCamData_.front();
                    m_QueCamData_.pop();
                }
                locker.unlock();

                //process  camera data//
                //01 单线程方式存数据
                SaveImage(tempdata);

                //02 多线程的方式存数据//
                m_Pool_->enqueue(t_SaveCameraImage, this, tempdata);
            }
        }
        catch (const std::exception& e)
        {
            std::cout << e.what() << "\n";
        }


        if (m_bQueExit_)
        {
            break;
        }
    }
}


void Sensor::SaveImage(std::shared_ptr<CameraData> data)
{
    //保存相机
    //do want you want
}

void t_SaveCameraImage(void* para, std::shared_ptr<CameraData> data)
{
    //保存相机数据
    //do want you want
}

作为数据采集程序，这段代码非常适用。任何传感器的数据我都能用一个容器先接着，然后另外一个线程处理从这个容器中弹出来的数据。如果采用回调的方式获取传感器数据，这种方式还不堵塞回调。既然作为一种通用的方式，这段代码当然可以用模板类进行改进。笔者尝试着写了一个模板类，目前测试有效，下一篇文章公布模板代码。

欢迎大牛们与我进行交流探讨，如果对上述代码有疑问或者缺少某个头文件（如SThreadPool，这是一个线程池模板类），请扫码加我微信，备注“C++”。

WOODS