如何用simdjson/rapidjson流式解析大体积json.gz文件？

你的场景完全可以通过simdjson的OnDemand模式或rapidjson的SAX模式解决，两者都支持增量/流式解析，无需一次性加载整个JSON文件。以下是针对两个库的具体实现方案：

一、使用simdjson的OnDemand模式（推荐，代码更简洁） #

simdjson的OnDemand API支持分块喂入JSON数据，自动处理未闭合的结构，无需手动维护解析上下文。你只需要缓存未完成的JSON片段，在下一次分块到来时继续解析。

修改后的完整代码：

#include <string>
#include <zlib.h>
#include <fstream>
#include <iostream>
#include "simdjson.h"

using namespace simdjson;
const int CHUNK_SIZE = 10240;

void decompress_and_parse(const std::string &filename) {
    gzFile gzFile = gzopen(filename.c_str(), "rb");
    if (!gzFile) {
        std::cerr << "打开压缩文件失败: " << filename << std::endl;
        return;
    }

    char buffer[CHUNK_SIZE];
    std::string leftover; // 缓存未完成的JSON片段
    ondemand::parser parser;
    ondemand::document doc;
    bool is_first_chunk = true;

    int bytesRead;
    while ((bytesRead = gzread(gzFile, buffer, sizeof(buffer))) > 0) {
        std::string chunk(buffer, bytesRead);
        leftover += chunk;

        try {
            // 初始化或继续解析缓存内容
            if (is_first_chunk) {
                doc = parser.iterate(leftover);
                is_first_chunk = false;
            } else {
                doc.continue_parsing(leftover);
            }

            // 这里根据你的JSON结构处理数据，示例为遍历顶级数组
            if (doc.is_array()) {
                for (auto element : doc.get_array()) {
                    // 替换为你的业务逻辑，比如提取字段
                    std::cout << "解析到元素: " << element.type() << std::endl;
                    // 示例：提取对象中的"id"字段
                    // if (element.is_object()) {
                    //     uint64_t id;
                    //     element["id"].get(id);
                    //     std::cout << "元素ID: " << id << std::endl;
                    // }
                }
            } else if (doc.is_object()) {
                // 处理顶级对象的逻辑
                auto obj = doc.get_object();
                for (auto field : obj) {
                    std::cout << "字段: " << field.key() << " 类型: " << field.value().type() << std::endl;
                }
            }

            // 解析完成后清空缓存
            leftover.clear();
        } catch (const simdjson_error &e) {
            // 仅当是"不完整JSON"错误时，保留缓存继续解析
            if (e.error_code() != error_code::INCOMPLETE_JSON) {
                std::cerr << "解析错误: " << e.what() << std::endl;
                break;
            }
        }
    }

    // 处理最后剩余的缓存内容
    if (!leftover.empty()) {
        try {
            if (is_first_chunk) {
                doc = parser.iterate(leftover);
            } else {
                doc.continue_parsing(leftover);
            }
            // 重复上述处理逻辑
            if (doc.is_array()) {
                for (auto element : doc.get_array()) {
                    std::cout << "最后解析到元素: " << element.type() << std::endl;
                }
            }
        } catch (const simdjson_error &e) {
            std::cerr << "剩余内容解析错误: " << e.what() << std::endl;
        }
    }

    if (bytesRead < 0) {
        std::cerr << "解压错误: " << gzerror(gzFile, NULL) << std::endl;
    }

    gzclose(gzFile);
}

int main() {
    auto start = std::chrono::high_resolution_clock::now();
    std::string filename = "data/example.json.gz";
    decompress_and_parse(filename);
    auto end = std::chrono::high_resolution_clock::now();
    std::chrono::duration<double> elapsed = end - start;
    std::cout << "\n耗时(秒): " << elapsed.count() << "\n" << std::endl;
    return 0;
}

二、使用rapidjson的SAX模式（内存占用更低） #

rapidjson的SAX API是纯流式解析，逐token处理JSON，无需缓存未完成片段，适合极端大文件场景。你需要实现自定义的SAX处理器，处理每个解析事件。

修改后的完整代码：

#include <string>
#include <zlib.h>
#include <fstream>
#include <iostream>
#include "rapidjson/reader.h"
#include "rapidjson/stringbuffer.h"

using namespace rapidjson;

// 自定义SAX处理器，实现你需要的解析逻辑
class MySaxHandler : public BaseReaderHandler<UTF8<>, MySaxHandler> {
public:
    bool Null() { std::cout << "解析到Null" << std::endl; return true; }
    bool Bool(bool b) { std::cout << "解析到Bool: " << (b ? "true" : "false") << std::endl; return true; }
    bool Int(int i) { std::cout << "解析到Int: " << i << std::endl; return true; }
    bool Uint(unsigned u) { std::cout << "解析到Uint: " << u << std::endl; return true; }
    bool Double(double d) { std::cout << "解析到Double: " << d << std::endl; return true; }
    bool String(const Ch* str, SizeType length, bool) { 
        std::cout << "解析到String: " << std::string(str, length) << std::endl; 
        return true; 
    }
    bool StartObject() { 
        std::cout << "开始解析Object" << std::endl; 
        object_depth++; 
        return true; 
    }
    bool Key(const Ch* str, SizeType length, bool) { 
        std::cout << "Object键: " << std::string(str, length) << std::endl; 
        return true; 
    }
    bool EndObject(SizeType memberCount) { 
        std::cout << "结束解析Object，成员数: " << memberCount << std::endl; 
        object_depth--; 
        return true; 
    }
    bool StartArray() { 
        std::cout << "开始解析Array" << std::endl; 
        array_depth++; 
        return true; 
    }
    bool EndArray(SizeType elementCount) { 
        std::cout << "结束解析Array，元素数: " << elementCount << std::endl; 
        array_depth--; 
        return true; 
    }

private:
    int object_depth = 0;
    int array_depth = 0;
};

const int CHUNK_SIZE = 10240;

void decompress_and_parse(const std::string &filename) {
    gzFile gzFile = gzopen(filename.c_str(), "rb");
    if (!gzFile) {
        std::cerr << "打开压缩文件失败: " << filename << std::endl;
        return;
    }

    char buffer[CHUNK_SIZE];
    Reader reader;
    MySaxHandler handler;
    reader.IterativeParseInit(); // 初始化迭代解析状态

    int bytesRead;
    while ((bytesRead = gzread(gzFile, buffer, sizeof(buffer))) > 0) {
        StringStream ss(buffer, bytesRead);
        // 分块解析，reader自动维护未完成的解析状态
        if (!reader.IterativeParseNext<UTF8<> >(ss, handler)) {
            if (reader.GetParseErrorCode() != kParseErrorIncompleteJSON) {
                std::cerr << "解析错误: " << GetParseError_En(reader.GetParseErrorCode()) << std::endl;
                break;
            }
            // 不完整JSON，继续读取下一块
        }
    }

    // 检查是否解析完成
    if (!reader.IterativeParseComplete()) {
        std::cerr << "最终解析不完整，可能存在JSON语法错误" << std::endl;
    }

    if (bytesRead < 0) {
        std::cerr << "解压错误: " << gzerror(gzFile, NULL) << std::endl;
    }

    gzclose(gzFile);
}

int main() {
    auto start = std::chrono::high_resolution_clock::now();
    std::string filename = "data/example.json.gz";
    decompress_and_parse(filename);
    auto end = std::chrono::high_resolution_clock::now();
    std::chrono::duration<double> elapsed = end - start;
    std::cout << "\n耗时(秒): " << elapsed.count() << "\n" << std::endl;
    return 0;
}

关键说明 #

• simdjson OnDemand：适合需要随机访问JSON元素的场景，代码简洁，无需手动维护解析状态，要求simdjson版本≥0.7.0。
• rapidjson SAX：内存占用极低，适合超大型文件，但需要手动实现所有解析回调，对JSON结构的控制更精细。
• 编译时需链接对应库：simdjson需添加-lsimdjson，rapidjson为头文件库，部分场景需链接-lrapidjson。

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