C++如何基于高速读取算法输出含完整行的文件块用于SQLite批量插入

核心思路 #

保留原有基于ifstream的带缓冲读取高性能逻辑，不做多余的文件流seek操作，在内存层增加残留半行拼接机制，严格保证每个输出分块都是完整行，单批大小稳定在10万行左右，性能和最优逐行读取方案基本持平。
禁止使用固定字节切割后回溯文件指针补全行的实现，这类方案会触发大量随机IO，大文件场景下性能会下降一个数量级以上

具体实现步骤 #

• 预设1MB~8MB的大尺寸读取缓冲（推荐4MB，适配大多数机械盘/SSD的块大小），每次从文件读取固定长度的字节块到内存，减少系统IO调用次数
• 维护一个全局leftover字符串，专门存储上一次读取块末尾没有找到换行符的半行残留
• 每次加载新的字节块后，先把leftover内容拼接到当前块头部，再清空leftover
• 遍历拼接后的当前块，逐次查找换行符：
  • 兼容\n（Unix格式）和\r\n（Windows格式）两种换行，遇到\r直接跳过即可
  • 每找到一个换行符，就提取两个换行符之间的内容作为完整行，存入当前批次集合，行计数+1
  • 当批次内行数达到10万阈值时，直接把整批数据交给SQLite插入逻辑处理，清空批次集合、重置计数
  • 遍历到当前块末尾时，把剩余没遇到换行符的内容全部存入leftover，留到下一次读取时拼接
• 文件读取到EOF后，检查leftover是否有剩余内容，有则作为最后一行加入最后一个批次，处理完所有批次后结束流程

核心代码实现 #

#include <fstream>
#include <string>
#include <vector>
#include <functional>
#include <stdexcept>

// 单批行数，可根据SQLite性能调整
constexpr size_t BATCH_ROW_NUM = 100000;
// 读取缓冲大小，4MB
constexpr size_t READ_BUF_SIZE = 4 * 1024 * 1024;

/**
 * 无截断大CSV分块读取
 * @param csvPath CSV文件路径
 * @param batchInserter 批量插入回调，入参为10万行左右的完整行集合，直接对接SQLite插入逻辑
 */
void readCsvByBatch(const std::string& csvPath,
                    const std::function<void(const std::vector<std::string>&)>& batchInserter) {
    std::ifstream csvFile(csvPath, std::ios::binary);
    if (!csvFile.is_open()) {
        throw std::runtime_error("CSV文件打开失败");
    }

    std::vector<char> readBuf(READ_BUF_SIZE);
    std::string leftover;
    std::vector<std::string> currentBatch;
    currentBatch.reserve(BATCH_ROW_NUM);

    while (csvFile.read(readBuf.data(), READ_BUF_SIZE) || csvFile.gcount() > 0) {
        size_t validLen = csvFile.gcount();
        // 拼接上一轮残留半行
        std::string currentBlock = std::move(leftover);
        currentBlock.append(readBuf.data(), validLen);
        leftover.clear();

        size_t lineStartPos = 0;
        for (size_t i = 0; i < currentBlock.size(); ++i) {
            if (currentBlock[i] == '\r') {
                continue;
            }
            if (currentBlock[i] == '\n') {
                // 提取完整行
                currentBatch.emplace_back(
                    currentBlock.substr(lineStartPos, i - lineStartPos)
                );
                lineStartPos = i + 1;

                // 达到批次大小，触发批量插入
                if (currentBatch.size() >= BATCH_ROW_NUM) {
                    batchInserter(currentBatch);
                    currentBatch.clear();
                }
            }
        }

        // 保存当前块末尾的半行残留
        if (lineStartPos < currentBlock.size()) {
            leftover = currentBlock.substr(lineStartPos);
        }
    }

    // 处理最后残留的行
    if (!leftover.empty()) {
        currentBatch.emplace_back(std::move(leftover));
    }
    // 处理最后一个不足10万行的批次
    if (!currentBatch.empty()) {
        batchInserter(currentBatch);
        currentBatch.clear();
    }

    csvFile.close();
}

常见问题排查&优化提示 #

• 之前测试代码仅输出固定值1，优先排查三个问题：
  • 没有用gcount()获取每次读取的有效字节数，直接遍历固定长度的缓冲数组，读到未初始化内存导致换行符判断失效
  • 换行符判断逻辑错误，比如把\r识别成行终止符，或者漏判了当前系统的换行格式
  • 批次计数变量作用域错误，达到第一次阈值后没有重置，导致后续无法继续收集行
• SQLite批量插入优化：拿到批次数据后，先用BEGIN TRANSACTION开启事务，预编译插入语句逐行绑定参数，整批插入完成后再COMMIT，10万行的插入耗时通常在几百毫秒级别，比逐行插入性能高2~3个数量级
• 代码中大量使用std::move转移字符串所有权，没有多余内存拷贝，处理10GB以上的超大CSV时，内存占用始终稳定在缓冲大小+单批数据的量级，不会出现内存暴涨的问题

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