Node.js中使用sax-js分块处理XML时异步批量入库丢数据问题

问题根源 #

你遇到的16条数据丢失，核心原因是：

• 调用xmlStream.pause()前，saxStream可能已经缓存了部分XML数据并解析出Item节点
• 原代码中只有documents.length < BATCH_SIZE时才会将Item加入数组，此时因为documents已满且isProcessing=true，这些解析出的Item直接被跳过，没有存入数组也没有入库
• 流结束时未处理剩余的Item（如果有的话）

修正方案 #

调整逻辑，确保所有Item都会被捕获，同时通过流暂停控制异步处理节奏，最后处理剩余数据：

const gunzip = zlib.createGunzip();
const xmlStream = fs.createReadStream(path).pipe(gunzip);
const saxStream = sax.createStream(true);
const BATCH_SIZE = 100;

let documents = [];
let currentElement = {};
let currentNode = null; 
let isProcessing = false;

saxStream.on("opentag", function (node) {
    currentNode = node.name;
    if (node.name === "Item") {
      currentElement = {}; // 初始化每个Item的对象
    }
});

saxStream.on("text", (text) => {
    if (currentElement) {
      doSomthing(text); // 确保此函数正确将text映射到currentElement的对应字段
    }
});

saxStream.on("closetag", async function (name) {
    if (name === "Item" && currentElement) {
      // 所有Item都先加入数组，不再判断数组大小
      documents.push(currentElement);
      currentElement = {};

      // 当数组达到批次大小且未在处理时，启动批量入库
      if (documents.length >= BATCH_SIZE && !isProcessing) {
        isProcessing = true;
        xmlStream.pause(); // 暂停数据流，避免继续解析新数据
        // 截取当前批次，剩余数据留在documents中
        const batch = [...documents];
        documents = [];
        
        console.log("Start process batch of size", batch.length);
        await insertDocuments(batch);
        console.log("End process batch of size", batch.length);
        
        isProcessing = false;
        xmlStream.resume(); // 恢复数据流，继续解析
      }
    }
    currentNode = null;
});

// 监听流结束事件，处理剩余的不足批次大小的数据
saxStream.on("end", async function() {
    if (documents.length > 0 && !isProcessing) {
      console.log("Start processing remaining batch of size", documents.length);
      await insertDocuments(documents);
      console.log("End processing remaining batch of size", documents.length);
      documents = [];
    }
});

xmlStream.pipe(saxStream);

关键改进点 #

• 移除数组大小判断：所有Item都会被加入documents，避免因批次处理中丢失新解析的节点
• 批次截取处理：处理时截取当前批次，剩余数据留在数组中，不影响后续新节点的加入
• 流结束收尾：确保最后一批不足BATCH_SIZE的数据也能被入库
• 严格流控制：暂停数据流后再处理批次，避免异步处理期间解析新数据导致的状态混乱

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