C++中无内存分配的字符串共享内存SPSC队列实现需求

我们可以基于环形缓冲区在共享内存中实现无额外内存分配的SPSC字符串队列，每个队列元素采用[uint32_t长度][字符串数据]的连续结构存储，完全避免Boost方案中字符串偏移和额外内存分配的开销。

共享内存布局 #

共享内存段分为控制块和数据缓冲区两部分，整体布局如下：

| 控制块（读写指针、缓冲区容量） | 长度 | 字符串1 | 长度 | 字符串2 | 长度 | 字符串3 | ... |

控制块管理队列的读写状态，数据缓冲区以环形方式连续存储所有字符串数据。

完整实现代码 #

共享内存控制结构 #

#include <cstdint>
#include <cstring>
#include <atomic>
#include <string>
#include <string_view>
#include <boost/interprocess/shared_memory_object.hpp>
#include <boost/interprocess/mapped_region.hpp>

namespace ipc = boost::interprocess;

// 共享内存中的控制块，用于管理环形缓冲区的读写状态
struct ShmQueueControl {
    std::atomic<uint64_t> write_pos = 0;  // 写入位置（字节偏移，相对于缓冲区起始）
    std::atomic<uint64_t> read_pos = 0;   // 读取位置（字节偏移）
    uint64_t buffer_size;                 // 环形缓冲区总容量（字节）
    char buffer[];                        // 柔性数组，作为环形缓冲区存储区
};

生产者（Writer）实现 #

class ShmQueueWriter {
private:
    ShmQueueControl* control_;

    // 计算环形缓冲区剩余可用空间
    uint64_t available_space() const {
        const uint64_t w = control_->write_pos.load(std::memory_order_acquire);
        const uint64_t r = control_->read_pos.load(std::memory_order_acquire);
        return (w >= r) 
            ? (control_->buffer_size - (w - r)) - sizeof(uint32_t) 
            : (r - w) - sizeof(uint32_t);
    }

public:
    explicit ShmQueueWriter(ShmQueueControl* control) : control_(control) {}

    // 写入字符串，空间不足时返回false
    bool write(std::string_view str) {
        const uint32_t str_size = static_cast<uint32_t>(str.size());
        const uint64_t required_space = sizeof(uint32_t) + str_size;

        // 检查是否有足够空间容纳当前字符串
        if (available_space() < required_space) {
            return false;
        }

        uint64_t w = control_->write_pos.load(std::memory_order_relaxed);
        char* buf = control_->buffer;

        // 写入字符串长度字段
        std::memcpy(buf + w, &str_size, sizeof(str_size));
        w += sizeof(str_size);

        // 处理环形绕回：数据跨缓冲区末尾时拆分写入
        if (w + str_size > control_->buffer_size) {
            const uint64_t first_segment = control_->buffer_size - w;
            std::memcpy(buf + w, str.data(), first_segment);
            std::memcpy(buf, str.data() + first_segment, str_size - first_segment);
            w = str_size - first_segment;
        } else {
            std::memcpy(buf + w, str.data(), str_size);
            w += str_size;
        }

        // 原子更新写入位置，确保消费者可见
        control_->write_pos.store(w, std::memory_order_release);
        return true;
    }
};

消费者（Reader）实现 #

class ShmQueueReader {
private:
    ShmQueueControl* control_;

public:
    explicit ShmQueueReader(ShmQueueControl* control) : control_(control) {}

    // 读取一个字符串到输出参数，队列为空时返回false
    bool read(std::string& out) {
        const uint64_t w = control_->write_pos.load(std::memory_order_acquire);
        const uint64_t r = control_->read_pos.load(std::memory_order_relaxed);
        
        if (w == r) {
            return false;  // 队列无数据
        }

        char* buf = control_->buffer;
        uint32_t str_size;

        // 读取字符串长度字段
        std::memcpy(&str_size, buf + r, sizeof(str_size));
        uint64_t next_r = r + sizeof(str_size);

        // 准备输出缓冲区
        out.resize(str_size);

        // 处理环形绕回：数据跨缓冲区末尾时拆分读取
        if (next_r + str_size > control_->buffer_size) {
            const uint64_t first_segment = control_->buffer_size - next_r;
            std::memcpy(out.data(), buf + next_r, first_segment);
            std::memcpy(out.data() + first_segment, buf, str_size - first_segment);
            next_r = str_size - first_segment;
        } else {
            std::memcpy(out.data(), buf + next_r, str_size);
            next_r += str_size;
        }

        // 原子更新读取位置，确保生产者可见
        control_->read_pos.store(next_r, std::memory_order_release);
        return true;
    }
};

使用示例 #

// 创建或打开共享内存对象
ipc::shared_memory_object shm(ipc::open_or_create, "SpscStringShmQueue", ipc::read_write);
const uint64_t buffer_capacity = 1024 * 1024;  // 1MB环形缓冲区容量
const uint64_t total_shm_size = sizeof(ShmQueueControl) + buffer_capacity;
shm.truncate(total_shm_size);

// 映射共享内存到当前进程地址空间
ipc::mapped_region region(shm, ipc::read_write);
ShmQueueControl* control = static_cast<ShmQueueControl*>(region.get_address());

// 首次创建共享内存时初始化缓冲区容量
if (control->buffer_size == 0) {
    control->buffer_size = buffer_capacity;
}

// 生产者端写入数据
ShmQueueWriter writer(control);
writer.write("Hello, Shared Memory SPSC Queue!");
writer.write("No extra memory allocation here!");

// 消费者端读取数据
ShmQueueReader reader(control);
std::string msg;
while (reader.read(msg)) {
    // 处理读取到的字符串
}

关键注意事项 #

• 原子操作与内存序：读写指针使用原子变量，acquire/release内存序保证跨进程的内存可见性，避免指令重排导致的状态不一致。
• 字符串长度限制：当前用uint32_t存储长度，支持最大4GB的单字符串；若场景中字符串更小，可改用uint16_t（最大64KB）减少内存开销。
• 对齐保证：memcpy操作天然处理数据对齐问题，控制块的成员布局由编译器自动保证对齐，无需额外处理。
• 空间与空队列检查：写入前必须检查可用空间，防止覆盖未读取的数据；读取前检查读写指针是否相等，判断队列是否为空。
• 异常退出处理：若生产者/消费者异常退出，可添加原子状态标志或版本号来重置队列状态，避免数据残留导致的异常。

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