如何在后台线程运行boost::iostreams::gzip_decompressor优化大JSON解析

在性能敏感场景下，我使用RapidJSON流式解析TB级的gzip压缩JSON文件。预解压文件再解析会产生显著延迟，因此尝试结合boost::iostreams::gzip_decompressor与filtering_istream实现流式解压+解析，代码如下：

ifstream jsonFile(filename, ios::binary);
boost::iostreams::filtering_istream gzipStream;
gzipStream.push(boost::iostreams::gzip_decompressor());
gzipStream.push(jsonFile);

char readBuffer[1048576];
rapidjson::IStreamWrapper iStreamWrapper(gzipStream, readBuffer, sizeof(readBuffer));

rapidjson::Reader reader;
ParseResult ok = reader.Parse(iStreamWrapper, handler);

该方案功能正常，但由于JSON解析与GUnzip均为CPU密集型任务，单线程执行导致性能下降约30%。希望将GUnzip放在后台线程，通过线程间流传递数据给解析线程，请问使用boost filtering_istream是否有合理的实现方式？

可以通过线程安全的生产者-消费者缓冲区结合自定义Boost.Iostreams源适配器，实现解压线程与解析线程的并行执行，具体实现步骤如下：

1. 实现线程安全的环形缓冲区 #

作为解压线程（生产者）和解析线程（消费者）的数据交换层，支持线程安全的读写与状态通知：

#include <boost/circular_buffer.hpp>
#include <mutex>
#include <condition_variable>

class ThreadSafeBuffer {
public:
    explicit ThreadSafeBuffer(size_t capacity) : buffer_(capacity) {}

    void push(const char* data, size_t size) {
        std::lock_guard<std::mutex> lock(mutex_);
        buffer_.insert(buffer_.end(), data, data + size);
        cond_.notify_one();
    }

    size_t pop(char* dest, size_t max_size) {
        std::unique_lock<std::mutex> lock(mutex_);
        cond_.wait(lock, [this] { return !buffer_.empty() || finished_; });
        
        size_t read_size = std::min(max_size, buffer_.size());
        std::copy(buffer_.begin(), buffer_.begin() + read_size, dest);
        buffer_.erase(buffer_.begin(), buffer_.begin() + read_size);
        return read_size;
    }

    void mark_finished() {
        std::lock_guard<std::mutex> lock(mutex_);
        finished_ = true;
        cond_.notify_one();
    }

private:
    boost::circular_buffer<char> buffer_;
    std::mutex mutex_;
    std::condition_variable cond_;
    bool finished_ = false;
};

2. 实现Boost.Iostreams的Source适配器 #

为线程安全缓冲区实现Source接口，让filtering_istream可以直接读取缓冲区数据：

#include <boost/iostreams/concepts.hpp>

class BufferSource : public boost::iostreams::source {
public:
    explicit BufferSource(ThreadSafeBuffer& buffer) : buffer_(buffer) {}

    std::streamsize read(char* s, std::streamsize n) override {
        size_t read = buffer_.pop(s, static_cast<size_t>(n));
        return read == 0 && buffer_finished_ ? -1 : static_cast<std::streamsize>(read);
    }

    void mark_finished() { buffer_finished_ = true; }

private:
    ThreadSafeBuffer& buffer_;
    bool buffer_finished_ = false;
};

3. 启动多线程执行流程 #

• 后台线程负责读取gzip文件、解压并写入缓冲区
• 主线程通过自定义Source读取数据，交给RapidJSON解析

#include <thread>
#include <fstream>
#include <boost/iostreams/filtering_stream.hpp>
#include <boost/iostreams/filter/gzip.hpp>
#include <rapidjson/reader.h>
#include <rapidjson/istreamwrapper.h>

void decompress_worker(const std::string& filename, ThreadSafeBuffer& buffer) {
    std::ifstream file(filename, std::ios::binary);
    boost::iostreams::filtering_istream decompressor;
    decompressor.push(boost::iostreams::gzip_decompressor());
    decompressor.push(file);

    char buffer_chunk[1048576];
    while (decompressor.read(buffer_chunk, sizeof(buffer_chunk))) {
        buffer.push(buffer_chunk, decompressor.gcount());
    }
    buffer.mark_finished();
}

int main() {
    const std::string filename = "large_data.json.gz";
    ThreadSafeBuffer ts_buffer(10 * 1048576); // 10MB环形缓冲区

    // 启动解压线程
    std::thread decompress_thread(decompress_worker, filename, std::ref(ts_buffer));

    // 构建解析用的流
    BufferSource source(ts_buffer);
    boost::iostreams::filtering_istream parse_stream;
    parse_stream.push(source);

    // RapidJSON解析逻辑
    char read_buffer[1048576];
    rapidjson::IStreamWrapper stream_wrapper(parse_stream, read_buffer, sizeof(read_buffer));
    rapidjson::Reader reader;
    CustomHandler handler; // 替换为你的自定义RapidJSON处理handler
    rapidjson::ParseResult ok = reader.Parse(stream_wrapper, handler);

    // 等待解压线程完成
    decompress_thread.join();
    source.mark_finished();

    if (!ok) {
        // 处理解析错误
        return 1;
    }
    return 0;
}

关键注意事项 #

• 缓冲区容量需根据内存情况调整：过大占用内存，过小会导致线程频繁阻塞
• 确保CustomHandler是线程安全的（若解析过程涉及共享状态修改）
• 可调整缓冲区块大小（示例为1MB）平衡线程切换开销与数据吞吐量

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