\u5728 Node.js \u4e2d\u5904\u7406\u5927\u6587\u4ef6\uff08\u5982\u51e0 GB \u7684\u65e5\u5fd7\u3001\u89c6\u9891\u6216\u6570\u636e\u96c6\uff09\u65f6\uff0c\u4f20\u7edf\u7684\u4e00\u6b21\u6027\u8bfb\u53d6\u6240\u6709\u5185\u5bb9\u5230\u5185\u5b58\u7684\u65b9\u5f0f\u4f1a\u8f7b\u6613\u8017\u5c3d\u5185\u5b58\uff0c\u751a\u81f3\u5bfc\u81f4\u8fdb\u7a0b\u5d29\u6e83\u3002Stream\uff08\u6d41\uff09\u6b63\u662f\u4e3a\u6b64\u800c\u751f\u2014\u2014\u5b83\u5c06\u6570\u636e\u5206\u5757\u5904\u7406\uff0c\u6bcf\u6b21\u53ea\u5360\u7528\u4e00\u5c0f\u5757\u5185\u5b58\uff0c\u4ece\u800c\u9ad8\u6548\u5904\u7406\u4efb\u610f\u5927\u5c0f\u7684\u6570\u636e\u3002<\/p>\n
\u7136\u800c\uff0c\u5f88\u591a\u5f00\u53d1\u8005\u5bf9 Stream \u7684\u8ba4\u77e5\u505c\u7559\u5728\u201c\u7528 pipe \u8fde\u63a5\u201d\u7684\u5c42\u9762\uff0c\u9047\u5230\u80cc\u538b\u3001\u9519\u8bef\u5904\u7406\u3001\u81ea\u5b9a\u4e49\u6d41\u7b49\u95ee\u9898\u65f6\u5e38\u5e38\u624b\u8db3\u65e0\u63aa\u3002\u672c\u6587\u5c06\u4ece\u5b9e\u6218\u51fa\u53d1\uff0c\u5e26\u4f60\u6df1\u5165\u7406\u89e3 Stream \u7684\u6838\u5fc3\u673a\u5236\uff0c\u5e76\u7ed9\u51fa\u53ef\u590d\u7528\u7684\u6700\u4f73\u5b9e\u8df5\u3002<\/p>\n
Node.js \u4e2d\u6709\u56db\u79cd\u57fa\u672c\u7684\u6d41\u7c7b\u578b\uff1a<\/p>\n
fs.createReadStream<\/code><\/li>\n- Writable<\/strong>\uff08\u53ef\u5199\u6d41\uff09\uff1a\u6570\u636e\u76ee\u6807\uff0c\u5982
fs.createWriteStream<\/code><\/li>\n- Transform<\/strong>\uff08\u8f6c\u6362\u6d41\uff09\uff1a\u65e2\u53ef\u8bfb\u53c8\u53ef\u5199\uff0c\u53ef\u5728\u8bfb\u5199\u8fc7\u7a0b\u4e2d\u4fee\u6539\u6570\u636e\uff0c\u5982
zlib.createGzip<\/code><\/li>\n- Duplex<\/strong>\uff08\u53cc\u5de5\u6d41\uff09\uff1a\u65e2\u53ef\u8bfb\u53c8\u53ef\u5199\uff0c\u4f46\u8bfb\u5199\u72ec\u7acb\uff0c\u5982
net.Socket<\/code><\/li>\n<\/ul>\n\u6838\u5fc3\u65b9\u6cd5\uff1a<\/p>\n
\nreadable.pipe(writable)<\/code>\uff1a\u5c06\u53ef\u8bfb\u6d41\u8fde\u63a5\u5230\u53ef\u5199\u6d41\uff0c\u81ea\u52a8\u5904\u7406\u80cc\u538b<\/li>\nreadable.on('data', chunk => ...)<\/code>\uff1a\u624b\u52a8\u6d88\u8d39\u6570\u636e\uff08\u4e0d\u63a8\u8350\uff0c\u9664\u975e\u9700\u8981\u7cbe\u7ec6\u63a7\u5236\uff09<\/li>\nwritable.write(chunk)<\/code>\uff1a\u5199\u5165\u6570\u636e\u5757<\/li>\nwritable.end()<\/code>\uff1a\u7ed3\u675f\u5199\u5165<\/li>\n<\/ul>\n\u5b9e\u6218\u4e00\uff1a\u5927\u6587\u4ef6\u590d\u5236\u4e0e\u80cc\u538b\u63a7\u5236<\/h2>\n\u9519\u8bef\u793a\u4f8b\uff1a\u4e00\u6b21\u6027\u8bfb\u53d6<\/h3>\n\nconst fs = require('fs');\n\n\/\/ \u5371\u9669\uff01\u5927\u6587\u4ef6\u4f1a\u8017\u5c3d\u5185\u5b58\nfs.readFile('bigfile.txt', (err, data) => {\n fs.writeFile('copy.txt', data, err => {\n if (err) console.error(err);\n });\n});\n<\/code><\/pre>\n\u6b63\u786e\u505a\u6cd5\uff1a\u4f7f\u7528 pipe \u81ea\u52a8\u80cc\u538b<\/h3>\n\nconst fs = require('fs');\n\nconst readable = fs.createReadStream('bigfile.txt');\nconst writable = fs.createWriteStream('copy.txt');\n\nreadable.pipe(writable);\n\nreadable.on('error', err => console.error('\u8bfb\u53d6\u9519\u8bef:', err));\nwritable.on('error', err => console.error('\u5199\u5165\u9519\u8bef:', err));\nwritable.on('finish', () => console.log('\u590d\u5236\u5b8c\u6210'));\n<\/code><\/pre>\n\u80cc\u538b\uff08Backpressure\uff09<\/strong>\uff1a\u5f53\u5199\u5165\u901f\u5ea6\u6162\u4e8e\u8bfb\u53d6\u901f\u5ea6\u65f6\uff0cpipe<\/code> \u4f1a\u81ea\u52a8\u6682\u505c\u8bfb\u53d6\uff0c\u7b49\u5f85\u5199\u5165\u5b8c\u6210\u540e\u518d\u7ee7\u7eed\u3002\u8fd9\u662f Stream \u6700\u5f3a\u5927\u7684\u7279\u6027\u4e4b\u4e00\u3002<\/p>\n\u624b\u52a8\u63a7\u5236\u80cc\u538b\uff08\u9ad8\u7ea7\u573a\u666f\uff09<\/h3>\n
\u6709\u65f6\u4f60\u9700\u8981\u66f4\u7cbe\u7ec6\u7684\u63a7\u5236\uff0c\u6bd4\u5982\u7edf\u8ba1\u8fdb\u5ea6\u6216\u8f6c\u6362\u6570\u636e\u3002\u6b64\u65f6\u53ef\u4ee5\u76d1\u542c drain<\/code> \u4e8b\u4ef6\uff1a<\/p>\n\nconst fs = require('fs');\n\nconst readable = fs.createReadStream('bigfile.txt');\nconst writable = fs.createWriteStream('copy.txt');\n\nlet totalBytes = 0;\n\nreadable.on('data', chunk => {\n totalBytes += chunk.length;\n \/\/ \u5982\u679c write \u8fd4\u56de false\uff0c\u8868\u793a\u7f13\u51b2\u533a\u5df2\u6ee1\uff0c\u5e94\u6682\u505c\u8bfb\u53d6\n if (!writable.write(chunk)) {\n readable.pause();\n }\n console.log(`\u5df2\u5904\u7406 ${totalBytes} \u5b57\u8282`);\n});\n\nwritable.on('drain', () => {\n readable.resume();\n});\n\nreadable.on('end', () => writable.end());\nreadable.on('error', err => console.error(err));\nwritable.on('finish', () => console.log('\u590d\u5236\u5b8c\u6210'));\n<\/code><\/pre>\n> \ud83d\udca1 \u6ce8\u610f<\/strong>\uff1a\u624b\u52a8\u5b9e\u73b0\u80cc\u538b\u65f6\uff0c\u52a1\u5fc5\u786e\u4fdd\u5728 drain<\/code> \u4e8b\u4ef6\u4e2d\u6062\u590d\u8bfb\u53d6\uff0c\u5426\u5219\u53ef\u80fd\u5bfc\u81f4\u5185\u5b58\u6cc4\u6f0f\u6216\u7a0b\u5e8f\u6302\u8d77\u3002<\/p>\n\u5b9e\u6218\u4e8c\uff1a\u81ea\u5b9a\u4e49 Transform \u6d41\u5b9e\u73b0\u6570\u636e\u8f6c\u6362<\/h2>\n
\u5047\u8bbe\u6211\u4eec\u9700\u8981\u5c06 CSV \u6587\u4ef6\u4e2d\u7684\u6bcf\u4e00\u884c\u8f6c\u6210 JSON \u5bf9\u8c61\uff0c\u5e76\u8fc7\u6ee4\u6389\u67d0\u4e9b\u5b57\u6bb5\u3002\u6211\u4eec\u53ef\u4ee5\u521b\u5efa\u4e00\u4e2a Transform \u6d41\uff1a<\/p>\n
\nconst { Transform } = require('stream');\nconst fs = require('fs');\n\nclass CsvToJsonTransform extends Transform {\n constructor(options = {}) {\n super(options);\n this.buffer = ''; \/\/ \u7528\u4e8e\u5b58\u50a8\u4e0d\u5b8c\u6574\u7684\u884c\n }\n\n _transform(chunk, encoding, callback) {\n this.buffer += chunk.toString();\n const lines = this.buffer.split('\\n');\n \/\/ \u4fdd\u7559\u6700\u540e\u4e00\u4e2a\u53ef\u80fd\u4e0d\u5b8c\u6574\u7684\u884c\n this.buffer = lines.pop();\n\n for (const line of lines) {\n if (line.trim() === '') continue;\n const [name, age, city] = line.split(',');\n const json = JSON.stringify({ name, age: parseInt(age), city }) + '\\n';\n this.push(json);\n }\n callback();\n }\n\n _flush(callback) {\n \/\/ \u5904\u7406\u6700\u540e\u5269\u4f59\u7684\u884c\n if (this.buffer.trim() !== '') {\n const [name, age, city] = this.buffer.split(',');\n const json = JSON.stringify({ name, age: parseInt(age), city }) + '\\n';\n this.push(json);\n }\n callback();\n }\n}\n\n\/\/ \u4f7f\u7528\nconst readable = fs.createReadStream('data.csv');\nconst writable = fs.createWriteStream('data.json');\nconst transformer = new CsvToJsonTransform();\n\nreadable.pipe(transformer).pipe(writable);\n\nreadable.on('error', err => console.error(err));\ntransformer.on('error', err => console.error(err));\nwritable.on('finish', () => console.log('\u8f6c\u6362\u5b8c\u6210'));\n<\/code><\/pre>\n> \ud83d\udca1 \u5173\u952e\u70b9<\/strong>\uff1a > – transform<\/code> \u65b9\u6cd5\u5904\u7406\u6bcf\u4e2a\u6570\u636e\u5757\uff0c\u6ce8\u610f\u5904\u7406\u8de8\u5757\u7684\u884c\uff08buffer \u673a\u5236\uff09 > – <\/em>flush<\/code> \u65b9\u6cd5\u5728\u6d41\u7ed3\u675f\u65f6\u88ab\u8c03\u7528\uff0c\u7528\u4e8e\u5904\u7406\u5269\u4f59\u6570\u636e > – \u4f7f\u7528 this.push()<\/code> \u8f93\u51fa\u8f6c\u6362\u540e\u7684\u6570\u636e<\/p>\n\u5b9e\u6218\u4e09\uff1a\u6570\u636e\u7ba1\u9053\u4e0e\u9519\u8bef\u5904\u7406\u6700\u4f73\u5b9e\u8df5<\/h2>\n
\u5b9e\u9645\u9879\u76ee\u4e2d\uff0c\u6211\u4eec\u7ecf\u5e38\u9700\u8981\u4e32\u8054\u591a\u4e2a\u6d41\uff08\u5982\u89e3\u538b\u3001\u89e3\u5bc6\u3001\u8f6c\u6362\u3001\u538b\u7f29\uff09\u3002\u9519\u8bef\u5904\u7406\u662f\u5bb9\u6613\u5ffd\u7565\u7684\u73af\u8282\uff0c\u4e00\u4e2a\u6d41\u7684\u9519\u8bef\u5982\u679c\u4e0d\u5904\u7406\uff0c\u4f1a\u4f20\u64ad\u5230\u6574\u4e2a\u7ba1\u9053\uff0c\u5bfc\u81f4\u7a0b\u5e8f\u5d29\u6e83\u3002<\/p>\n
\u7ba1\u9053\u4e2d\u7684\u9519\u8bef\u4f20\u64ad<\/h3>\n\nconst fs = require('fs');\nconst zlib = require('zlib');\nconst { pipeline } = require('stream');\n\nconst readable = fs.createReadStream('archive.gz');\nconst gunzip = zlib.createGunzip();\nconst writable = fs.createWriteStream('output.txt');\n\n\/\/ \u4f7f\u7528 pipeline \u800c\u4e0d\u662f pipe\uff0c\u81ea\u52a8\u5904\u7406\u9519\u8bef\u548c\u6e05\u7406\npipeline(\n readable,\n gunzip,\n writable,\n (err) => {\n if (err) {\n console.error('Pipeline \u5931\u8d25:', err);\n } else {\n console.log('Pipeline \u6210\u529f');\n }\n }\n);\n<\/code><\/pre>\n\u4e3a\u4ec0\u4e48\u63a8\u8350 pipeline<\/code>\uff1f<\/strong><\/p>\n\npipe<\/code> \u4e0d\u4f1a\u81ea\u52a8\u9500\u6bc1\u6d41\uff0c\u9519\u8bef\u53ef\u80fd\u5bfc\u81f4\u5185\u5b58\u6cc4\u6f0f<\/li>\npipeline<\/code> \u5728 Node.js 10+ \u4e2d\u5f15\u5165\uff0c\u4f1a\u81ea\u52a8\u9500\u6bc1\u6240\u6709\u6d41\u5e76\u8c03\u7528\u56de\u8c03<\/li>\n- \u624b\u52a8\u4f7f\u7528
pipe<\/code> \u9700\u8981\u76d1\u542c\u6bcf\u4e2a\u6d41\u7684\u9519\u8bef\u5e76\u624b\u52a8\u9500\u6bc1\uff0c\u975e\u5e38\u7e41\u7410<\/li>\n<\/ul>\n\u81ea\u5b9a\u4e49\u9519\u8bef\u5904\u7406\u4e0e\u91cd\u8bd5<\/h3>\n
\u5bf9\u4e8e\u7f51\u7edc\u8bf7\u6c42\u7b49\u573a\u666f\uff0c\u6211\u4eec\u53ef\u4ee5\u5c01\u88c5\u4e00\u4e2a\u53ef\u91cd\u8bd5\u7684\u6d41\uff1a<\/p>\n
\nconst { Transform } = require('stream');\n\nclass RetryTransform extends Transform {\n constructor(options = {}) {\n super(options);\n this.retries = options.retries || 3;\n }\n\n _transform(chunk, encoding, callback) {\n \/\/ \u6a21\u62df\u4e00\u4e2a\u53ef\u80fd\u5931\u8d25\u7684\u64cd\u4f5c\n this.processWithRetry(chunk, 0, callback);\n }\n\n processWithRetry(chunk, attempt, callback) {\n try {\n \/\/ \u5047\u8bbe\u8fd9\u91cc\u53ef\u80fd\u629b\u51fa\u9519\u8bef\n if (Math.random() < 0.3) throw new Error('\u968f\u673a\u5931\u8d25');\n const result = chunk.toString().toUpperCase();\n this.push(result);\n callback();\n } catch (err) {\n if (attempt < this.retries) {\n console.log(`\u91cd\u8bd5\u7b2c ${attempt + 1} \u6b21`);\n setTimeout(() => this.processWithRetry(chunk, attempt + 1, callback), 1000);\n } else {\n callback(err); \/\/ \u8d85\u8fc7\u91cd\u8bd5\u6b21\u6570\uff0c\u4f20\u9012\u9519\u8bef\n }\n }\n }\n}\n<\/code><\/pre>\n> \u26a0\ufe0f \u6ce8\u610f<\/strong>\uff1a\u91cd\u8bd5\u903b\u8f91\u5e94\u8c28\u614e\u4f7f\u7528\uff0c\u907f\u514d\u65e0\u9650\u91cd\u8bd5\u5bfc\u81f4\u8d44\u6e90\u8017\u5c3d\u3002<\/p>\n\u5b9e\u6218\u56db\uff1a\u5bf9\u8c61\u6a21\u5f0f\u6d41\uff08Object Mode\uff09<\/h2>\n
\u9ed8\u8ba4\u60c5\u51b5\u4e0b\uff0c\u6d41\u5904\u7406\u7684\u662f Buffer \u6216\u5b57\u7b26\u4e32\u3002\u4f46\u6709\u65f6\u6211\u4eec\u9700\u8981\u5904\u7406 JavaScript \u5bf9\u8c61\uff0c\u8fd9\u65f6\u9700\u8981\u542f\u7528\u5bf9\u8c61\u6a21\u5f0f\uff1a<\/p>\n
\nconst { Transform } = require('stream');\n\nconst objectStream = new Transform({\n objectMode: true,\n transform(chunk, encoding, callback) {\n \/\/ chunk \u662f\u4e00\u4e2a\u5bf9\u8c61\n chunk.timestamp = Date.now();\n this.push(chunk);\n callback();\n }\n});\n\nobjectStream.write({ name: 'Alice' });\nobjectStream.on('data', obj => console.log(obj));\n\/\/ \u8f93\u51fa: { name: 'Alice', timestamp: 1700000000000 }\n<\/code><\/pre>\n> \ud83d\udca1 \u6ce8\u610f<\/strong>\uff1a\u5bf9\u8c61\u6a21\u5f0f\u4f1a\u5927\u5e45\u964d\u4f4e\u6027\u80fd\uff0c\u56e0\u4e3a\u6bcf\u4e2a\u5bf9\u8c61\u90fd\u662f\u4e00\u4e2a\u72ec\u7acb\u7684\u6570\u636e\u5757\uff0c\u5931\u53bb\u4e86 Buffer \u7684\u6279\u5904\u7406\u4f18\u52bf\u3002\u4ec5\u5f53\u9700\u8981\u5904\u7406\u7ed3\u6784\u5316\u6570\u636e\u4e14\u6570\u636e\u91cf\u4e0d\u5927\u65f6\u4f7f\u7528\u3002<\/p>\n\u6027\u80fd\u4f18\u5316\u4e0e\u76d1\u63a7<\/h2>\n\u9ad8\u6c34\u4f4d\u7ebf\uff08HighWaterMark\uff09<\/h3>\n
highWaterMark<\/code> \u63a7\u5236\u6d41\u5185\u90e8\u7f13\u51b2\u533a\u7684\u6700\u5927\u5b57\u8282\u6570\uff08\u6216\u5bf9\u8c61\u6570\uff09\u3002\u8c03\u6574\u5b83\u53ef\u4ee5\u5f71\u54cd\u80cc\u538b\u884c\u4e3a\uff1a<\/p>\n\n\/\/ \u589e\u5927\u7f13\u51b2\u533a\u4ee5\u51cf\u5c11\u80cc\u538b\u89e6\u53d1\u9891\u7387\uff08\u9002\u5408\u9ad8\u901f\u8bfb\u53d6\uff09\nconst readable = fs.createReadStream('bigfile.txt', { highWaterMark: 1024 * 1024 }); \/\/ 1MB\n\/\/ \u51cf\u5c0f\u7f13\u51b2\u533a\u4ee5\u964d\u4f4e\u5185\u5b58\u5360\u7528\uff08\u9002\u5408\u4f4e\u901f\u5199\u5165\uff09\nconst writable = fs.createWriteStream('output.txt', { highWaterMark: 16 * 1024 }); \/\/ 16KB\n<\/code><\/pre>\n\u4f7f\u7528 perf_hooks \u76d1\u63a7\u6d41\u6027\u80fd<\/h3>\n\nconst { performance, PerformanceObserver } = require('perf_hooks');\nconst fs = require('fs');\n\nconst obs = new PerformanceObserver((items) => {\n console.log(items.getEntries()[0]);\n});\nobs.observe({ entryTypes: ['measure'] });\n\nperformance.mark('start');\n\nconst readable = fs.createReadStream('bigfile.txt');\nconst writable = fs.createWriteStream('copy.txt');\n\nreadable.pipe(writable);\n\nwritable.on('finish', () => {\n performance.mark('end');\n performance.measure('copy', 'start', 'end');\n});\n<\/code><\/pre>\n\u603b\u7ed3<\/h2>\n\n- \u6c38\u8fdc\u4f7f\u7528 Stream \u5904\u7406\u5927\u6587\u4ef6<\/strong>\uff0c\u907f\u514d
readFile<\/code>\/writeFile<\/code><\/li>\n- \u4f18\u5148\u4f7f\u7528
pipeline<\/code> \u800c\u4e0d\u662f pipe<\/code><\/strong>\uff0c\u81ea\u52a8\u5904\u7406\u9519\u8bef\u548c\u6e05\u7406<\/li>\n- \u81ea\u5b9a\u4e49 Transform \u6d41\u65f6\uff0c\u6ce8\u610f\u8de8\u5757\u6570\u636e\u5904\u7406\u548c
_flush<\/code><\/strong><\/li>\n- \u7406\u89e3\u80cc\u538b\u673a\u5236<\/strong>\uff0c\u624b\u52a8\u63a7\u5236\u65f6\u52a1\u5fc5\u76d1\u542c
drain<\/code> \u4e8b\u4ef6<\/li>\n- \u5bf9\u8c61\u6a21\u5f0f\u53ea\u5728\u5c0f\u6570\u636e\u91cf\u65f6\u4f7f\u7528<\/strong>\uff0c\u6ce8\u610f\u6027\u80fd\u5f00\u9500<\/li>\n
- \u8c03\u6574
highWaterMark<\/code><\/strong> \u53ef\u4ee5\u4f18\u5316\u7279\u5b9a\u573a\u666f\u7684\u6027\u80fd<\/li>\n<\/ul>\n\u5ef6\u4f38\u9605\u8bfb<\/h2>\n