js Stream API简单使用

ReadableStream

<script>// TextDecoder将字节转换为字符串,默认 utf-8 编码let uint8Array = new Uint8Array([72, 101, 108, 108, 111]);console.log(new TextDecoder().decode(uint8Array)); // Hellolet uint8Array1 = new Uint8Array([228, 189, 160, 229, 165, 189]);console.log(new TextDecoder().decode(uint8Array1)); // 你好let uint8Array2 = new Uint8Array([0, 72, 101, 108, 108, 111, 0]);let binaryString = uint8Array2.subarray(1, -1);console.log(new TextDecoder().decode(binaryString)); // Hello// TextEncoder将字符串转换为字节console.log(new TextEncoder().encode("Hello")); // Uint8Array 72,101,108,108,111//ReadableStream 可读取的二进制流// 数据被按序读入到许多小的片段，这些片段被称作分块（chunk）。分块可以是单个字节，也可以是某种更大的数据类型，例如特定大小的类型化数组。单个流的分块可以有不同的大小和类型。// 已放入到流中的分块称作已入队（enqueued）——这意味着它们已经在队列中排队等待被读取。流的一个内置队列跟踪了那些尚未读取的分块。// 流中的分块由一个 reader 读取——该数据处理过程一次只处理一个分块，允许你对其执行任何类型的操作。// 每个 reader 都有一个关联的 controller，用来控制流例如可以将流关闭。fetch("http://localhost:3000/html-demos/table.html").then((response) => response.body).then((rb) => {const reader = rb.getReader();return new ReadableStream({start(controller) {// 处理每一个数据块function push() {reader.read().then(({ done, value }) => {// 读取完全部数据块if (done) {console.log("done", done);controller.close();return;}// 获取数据value，并通过controller发送给浏览器controller.enqueue(value);console.log(done, value);push();});}push();},});}).then((stream) =>// 处理流数据new Response(stream, { headers: { "Content-Type": "text/html" } }).text()).then((result) => {console.log(result);});
</script>

WritableStream

<script> const list = document.querySelector("ul");function sendMessage(message, writableStream) {const defaultWriter = writableStream.getWriter();const encoder = new TextEncoder();const encoded = encoder.encode(message, { stream: true });encoded.forEach((chunk) => {defaultWriter.ready.then(() => {return defaultWriter.write(chunk);}).then(() => {console.log(chunk, "Chunk written to sink.");}).catch((err) => {console.log("Chunk error:", err);});});// 重新调用ready确保writer关闭前所有数据已写入defaultWriter.ready.then(() => {defaultWriter.close();}).then(() => {console.log("All chunks written");}).catch((err) => {console.log("Stream error:", err);});}const decoder = new TextDecoder("utf-8");// CountQueuingStrategy 接口提供了一个内置的、用于对分块进行计数的队列策略，可以在构造流的时候使用。const queuingStrategy = new CountQueuingStrategy({ highWaterMark: 1 });let result = "";// 一个可写流（Writable stream）是一个可以写入数据的数据终点const writableStream = new WritableStream({write(chunk) {return new Promise((resolve, reject) => {var buffer = new ArrayBuffer(1);var view = new Uint8Array(buffer);view[0] = chunk;var decoded = decoder.decode(view, { stream: true });var listItem = document.createElement("li");listItem.textContent = "Chunk decoded: " + decoded;list.appendChild(listItem);result += decoded;resolve();});},close() {var listItem = document.createElement("li");listItem.textContent = "[MESSAGE RECEIVED] " + result;list.appendChild(listItem);},abort(err) {console.log("Sink error:", err);},},queuingStrategy);sendMessage("Hello, world.", writableStream);
</script>