跨类环形缓冲区数据交互及多线程管理问题求助

问题描述 #

我需要实现一个环形缓冲区，由readSensor类向其中写入数据，processData类负责创建缓冲区、读取数据并删除节点。最初在同一类中实现时功能正常，但跨类读取缓冲区失败，排查后发现是重复创建processData实例导致缓冲区地址不同；改用已有实例引用后功能恢复，但仍希望找到更优的多线程管理方案。

完整代码 #

testfile

#include <iostream>
#include <thread>
#include <mutex>
#include <chrono>
#include <string>

#include "buffer.hpp"


std::mutex bufferMutex;


struct TEST_DATA
{
double val_a;
double val_b;
double val_c;
};


class processData
{
public:
processData() {}
~processData() {}

CircularBuffer<TEST_DATA>& getTESTBuffer() { return test_buffer; 
}

void start()
{
    create();
    read();
}

void create()
{
    std::lock_guard<std::mutex> lock(bufferMutex);
    std::cout << "\nCreate buffer";
    test_buffer.createBuffer();
    printf("\nFrom processData: ");
    std::cout << "\nAddress of test_buffer: " << &test_buffer << std::endl;
}

void read()
{
    printf("\nFrom processData: ");
    TEST_DATA read_data;
    std::cout << "\nPM: Trying to read from Buffer\n";
    for(;;)
    { 
        while (test_buffer.readFromBuffer(read_data))
            std::cout << "Val a: " << read_data.val_a << " Val b: " << read_data.val_b << std::endl;
    }
}

private:
CircularBuffer<TEST_DATA> test_buffer;
 };



class readSensor
{
public:
readSensor(processData& sensor, std::mutex& mutex) : sensor(sensor), bufferMutex(mutex) {}
~readSensor() {}

void start()
{
    printf("\nFrom readSensor: before mutex, wait... ");
    std::lock_guard<std::mutex> lock(bufferMutex);
    printf("\nFrom readSensor: ");
    CircularBuffer<TEST_DATA>& bufferRef = sensor.getTESTBuffer();
    std::cout << "\nAddress of the buffer: " << &bufferRef << std::endl;
    std::cout << "Write to buffer" << std::endl;
    TEST_DATA test_data1;
    test_data1.val_a = 1234;
    test_data1.val_b = 9876;
    bufferRef.writeToBuffer(test_data1);

    std::this_thread::sleep_for(std::chrono::seconds(5));

    TEST_DATA test_data2;
    test_data2.val_a = 9999;
    test_data2.val_b = 8888;
    bufferRef.writeToBuffer(test_data2);
}

private:
processData& sensor;
std::mutex& bufferMutex;

};



int main()
{
processData sensor;
readSensor reader(sensor, bufferMutex);

std::thread processData_thread([&sensor]() { sensor.start(); });
std::this_thread::sleep_for(std::chrono::milliseconds(100));
std::thread readSensor_thread([&reader]() { reader.start(); });

readSensor_thread.join();
processData_thread.join();

return EXIT_SUCCESS;
}

buffer.hpp

#ifndef BUFFER_HPP_INCLUDED
#define BUFFER_HPP_INCLUDED

#include <string>
#include <mutex>

template<typename T>
struct CircularNode
{
T data;
CircularNode* next;
};

template<typename T>
class CircularBuffer
{
public:
CircularBuffer() : head(NULL), tail(NULL), buffer_created(false),     
node_count(0) {}
~CircularBuffer() {}

bool createBuffer()
{
    if (buffer_created)
        return false;
    head = NULL;
    tail = NULL;
    buffer_created = true;
    node_count = 0;
    return true;
}

bool writeToBuffer(const T& data)
{
    std::lock_guard<std::mutex> lock(buffer_mutex);
    if (!buffer_created)
        return false;
    try
    {
        CircularNode<T>* new_node = new CircularNode<T>{ data, 
   nullptr };
        if (head == nullptr)
        {
            head = new_node;
            tail = new_node;
            new_node->next = new_node;
        }
        else
        {
            tail->next = new_node;
            tail = new_node;
            tail->next = head;
        }
        ++node_count;
        return true;
    }
    catch (const std::bad_alloc& e)
    {
        return false;
    }
}

bool readFromBuffer(T& out_data)
{
    std::lock_guard<std::mutex> lock(buffer_mutex);

    if (head != NULL)
    {
        out_data = head->data;
        CircularNode<T>* temp = head;
        head = head->next;
        if (head == temp)
        {
            head = NULL;
            tail = NULL;
        }
        else
            tail->next = head;
        delete temp;
        --node_count;
        return true;
    }
    return false;
}

private:
CircularNode<T>* head;
CircularNode<T>* tail;
mutable std::mutex buffer_mutex;
bool buffer_created;
std::string buffer_name;
size_t node_count;
};

#endif

优化方案 #

1. 移除全局冗余互斥量：CircularBuffer内部已经实现了buffer_mutex用于读写同步，全局的bufferMutex完全多余，会增加不必要的锁竞争。直接删除全局互斥量，移除processData::create和readSensor::start中的相关锁逻辑即可。

2. 替换线程等待方式：main中用固定时长睡眠保证缓冲区初始化的方式不可靠，改用std::condition_variable实现同步：在processData中添加条件变量，缓冲区初始化完成后主动通知readSensor线程，避免依赖不确定的睡眠时长。

3. 优化读取线程空转问题：processData::read中的无限循环在缓冲区为空时会持续消耗CPU，结合std::condition_variable让读取线程在缓冲区为空时进入等待状态，有新数据写入时再唤醒，提升资源利用率。

4. 完善缓冲区析构逻辑：当前CircularBuffer析构时未清理剩余节点，会导致内存泄漏。在析构函数中遍历所有节点并释放内存。

5. 封装线程管理：将线程的创建、启动、销毁逻辑封装到类内部，比如让processData和readSensor各自持有线程对象，对外提供start()和stop()接口，简化主线程的代码结构。

内容的提问来源于stack exchange，提问作者michael86