Streams API 管道化处理与持久化实践

`title: Streams API 管道化处理与持久化实践``categories: Web 开发/前端/数据管理``keywords: Streams API,ReadableStream,WritableStream,TransformStream,OPFS,IndexedDB``description: 使用 Streams API 实现管道化数据处理与非阻塞持久化，结合 OPFS 与 IndexedDB 在浏览器端高效管理大数据。`适用场景大文本与二进制分块处理与写入前端生成数据的流水线化转换与持久化基本管道const upper = new TransformStream({

transform(chunk, controller) {

const s = new TextDecoder().decode(chunk);

controller.enqueue(new TextEncoder().encode(s.toUpperCase()));

}

});

async function pipeTextToOPFS(name, text) {

const root = await navigator.storage.getDirectory();

const handle = await root.getFileHandle(name, { create: true });

const writable = await handle.createWritable();

const src = new ReadableStream({

start(controller) {

controller.enqueue(new TextEncoder().encode(text));

controller.close();

}

});

await src.pipeThrough(upper).pipeTo(writable);

}

从网络到持久化async function saveResponseToOPFS(url, name) {

const res = await fetch(url);

const root = await navigator.storage.getDirectory();

const handle = await root.getFileHandle(name, { create: true });

const writable = await handle.createWritable();

await res.body.pipeTo(writable);

}

搭配 IndexedDBfunction openDB() {

return new Promise((resolve, reject) => {

const req = indexedDB.open('streams-db', 1);

req.onupgradeneeded = () => {

const db = req.result;

if (!db.objectStoreNames.contains('meta')) db.createObjectStore('meta', { keyPath: 'id' });

};

req.onsuccess = () => resolve(req.result);

req.onerror = () => reject(req.error);

});

}

async function saveMeta(db, record) {

return new Promise((resolve, reject) => {

const tx = db.transaction('meta', 'readwrite');

tx.objectStore('meta').put(record);

tx.oncomplete = () => resolve();

tx.onerror = () => reject(tx.error);

});

}