NDJSON 流式解析与增量持久化实践

概述NDJSON 适合流式处理。本文通过 `TextDecoderStream + TransformStream` 将文本流拆分为对象并增量持久化。能力检测与解析管道const supportsStreams = typeof ReadableStream === 'function' && typeof TransformStream === 'function'; function lineSplitter() { let buf = ''; return new TransformStream({ transform(chunk, controller) { buf += chunk; const parts = buf.split('\n'); buf = parts.pop(); for (const p of parts) if (p.trim()) controller.enqueue(JSON.parse(p)); }, flush(controller) { if (buf.trim()) controller.enqueue(JSON.parse(buf)); } }); } 增量持久化到 IndexedDBfunction openDB(name) { return new Promise((resolve, reject) => { const r = indexedDB.open(name, 1); r.onupgradeneeded = () => { const db = r.result; if (!db.objectStoreNames.contains('items')) db.createObjectStore('items', { keyPath: 'id' }); }; r.onsuccess = () => resolve(r.result); r.onerror = () => reject(r.error); }); } async function importNDJSON(url) { if (!supportsStreams) throw new Error('streams unsupported'); const res = await fetch(url); const rs = res.body.pipeThrough(new TextDecoderStream()).pipeThrough(lineSplitter()); const reader = rs.getReader(); const db = await openDB('ndjson'); let count = 0; while (true) { const { value, done } = await reader.read(); if (done) break; const tx = db.transaction('items', 'readwrite'); tx.objectStore('items').put(value); await new Promise((resolve, reject) => { tx.oncomplete = resolve; tx.onerror = () => reject(tx.error); }); count++; } db.close(); return count; }