\u5728 Node.js \u5f00\u53d1\u4e2d\uff0c\u5904\u7406\u5927\u6587\u4ef6\u6216\u5927\u91cf\u6570\u636e\u65f6\uff0c\u4f20\u7edf\u7684\u4e00\u6b21\u6027\u8bfb\u53d6\u65b9\u5f0f\uff08\u5982 Node.js \u7684\u6d41\u6709\u56db\u79cd\u7c7b\u578b\uff1a<\/p>\n \u6240\u6709\u6d41\u90fd\u662f EventEmitter \u7684\u5b9e\u4f8b\uff0c\u901a\u8fc7\u4e8b\u4ef6\u9a71\u52a8\u3002<\/p>\n \u6ce8\u610f<\/strong>\uff1a\u624b\u52a8\u7ba1\u7406\u8bfb\u5199\u5bb9\u6613\u51fa\u9519\uff0c\u6bd4\u5982\u5fd8\u8bb0\u8c03\u7528 \u7ba1\u9053\u53ef\u4ee5\u94fe\u5f0f\u8c03\u7528\uff0c\u5f62\u6210\u6570\u636e\u5904\u7406\u6d41\u6c34\u7ebf\u3002<\/p>\n \ud83d\udca1 \u6700\u4f73\u5b9e\u8df5<\/strong>\uff1a\u59cb\u7ec8\u5728\u6700\u540e\u4e00\u4e2a\u6d41\u4e0a\u76d1\u542c \u8f6c\u6362\u6d41\u7528\u4e8e\u5728\u6570\u636e\u4f20\u9012\u8fc7\u7a0b\u4e2d\u4fee\u6539\u6570\u636e\u3002\u4f8b\u5982\uff0c\u521b\u5efa\u4e00\u4e2a\u5c06\u6587\u672c\u8f6c\u4e3a\u5927\u5199\u7684\u6d41\u3002<\/p>\n \u5047\u8bbe\u6211\u4eec\u9700\u8981\u8bfb\u53d6\u4e00\u4e2a\u5927\u578b CSV \u6587\u4ef6\uff0c\u8fc7\u6ee4\u6389\u67d0\u4e9b\u884c\uff0c\u5e76\u8f93\u51fa\u4e3a JSON\u3002<\/p>\n \u6ce8\u610f<\/strong>\uff1a\u4f7f\u7528 \u80cc\u538b\u662f\u6307\u5f53\u5199\u5165\u901f\u5ea6\u6162\u4e8e\u8bfb\u53d6\u901f\u5ea6\u65f6\uff0c\u6d41\u81ea\u52a8\u63a7\u5236\u6570\u636e\u6d41\u52a8\u7684\u673a\u5236\u3002 \ud83d\udca1 \u6700\u4f73\u5b9e\u8df5<\/strong>\uff1a\u9664\u975e\u6709\u7279\u6b8a\u9700\u6c42\uff0c\u5426\u5219\u59cb\u7ec8\u4f7f\u7528 \u5904\u7406\u5927\u6587\u4ef6\u65f6\uff0c\u9010\u884c\u8bfb\u53d6\u6bd4\u6309 chunk \u8bfb\u53d6\u66f4\u65b9\u4fbf\u3002\u4f46 Node.js \u6ca1\u6709\u5185\u7f6e\u7684\u884c\u8bfb\u53d6\u6d41\uff0c\u6211\u4eec\u53ef\u4ee5\u81ea\u5df1\u5b9e\u73b0\u6216\u4f7f\u7528 \u6ce8\u610f<\/strong>\uff1a \u6d41\u4e8b\u4ef6\u4e2d\u7684\u9519\u8bef\u5fc5\u987b\u88ab\u6355\u83b7\uff0c\u5426\u5219\u4f1a\u5bfc\u81f4\u8fdb\u7a0b\u5d29\u6e83\u3002<\/p>\n \u540c\u6b65\u64cd\u4f5c\u4f1a\u963b\u585e\u4e8b\u4ef6\u5faa\u73af\uff0c\u5f71\u54cd\u6027\u80fd\u3002<\/p>\n \u6839\u636e\u5185\u5b58\u548c\u6027\u80fd\u9700\u6c42\u8c03\u6574 \u5bf9\u4e8e\u5c0f\u6587\u4ef6\uff08< 50MB\uff09\uff0c\u76f4\u63a5\u4f7f\u7528 Node.js Stream \u662f\u5904\u7406\u5927\u6587\u4ef6\u548c\u6784\u5efa\u6570\u636e\u7ba1\u9053\u7684\u5229\u5668\u3002\u901a\u8fc7 \u5ef6\u4f38\u9605\u8bfb<\/strong>\uff1a<\/p>\n \u672c\u6587\u6df1\u5165\u8bb2\u89e3 Node.js Stream \u7684\u6838\u5fc3\u6982\u5ff5\uff0c\u5e76\u901a\u8fc7\u5b9e\u9645\u6848\u4f8b\u6f14\u793a\u5982\u4f55\u4f7f\u7528\u6d41\u5904\u7406\u5927\u6587\u4ef6\u3001\u6784\u5efa\u6570\u636e\u7ba1\u9053\uff0c\u4ee5\u53ca\u907f\u514d\u5e38\u89c1\u9677\u9631\uff0c\u5e2e\u52a9\u8bfb\u8005\u63d0\u5347 Node.js \u6570\u636e\u5904\u7406\u80fd\u529b\u3002<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[3,4],"tags":[5,72,73,143,75,74],"class_list":["post-218","post","type-post","status-publish","format-standard","hentry","category-3","category-4","tag-node-js","tag-stream","tag-73","tag-143","tag-75","tag-74"],"_links":{"self":[{"href":"https:\/\/yykcj.com\/index.php?rest_route=\/wp\/v2\/posts\/218","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/yykcj.com\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/yykcj.com\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/yykcj.com\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/yykcj.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=218"}],"version-history":[{"count":0,"href":"https:\/\/yykcj.com\/index.php?rest_route=\/wp\/v2\/posts\/218\/revisions"}],"wp:attachment":[{"href":"https:\/\/yykcj.com\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=218"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/yykcj.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=218"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/yykcj.com\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=218"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}fs.readFile<\/code>\uff09\u5f80\u5f80\u4f1a\u5bfc\u81f4\u5185\u5b58\u6ea2\u51fa\u6216\u6027\u80fd\u74f6\u9888\u3002\u6d41\uff08Stream\uff09\u662f Node.js \u4e2d\u5904\u7406\u6570\u636e\u7684\u6838\u5fc3\u62bd\u8c61\uff0c\u5b83\u5141\u8bb8\u4f60\u4ee5\u5206\u5757\u7684\u65b9\u5f0f\u5904\u7406\u6570\u636e\uff0c\u4ece\u800c\u8282\u7701\u5185\u5b58\u5e76\u63d0\u9ad8\u6548\u7387\u3002\u672c\u6587\u5c06\u5e26\u4f60\u4ece\u96f6\u5f00\u59cb\uff0c\u638c\u63e1 Node.js Stream \u7684\u5b9e\u6218\u6280\u5de7\uff0c\u5305\u62ec\u8bfb\u5199\u6d41\u3001\u8f6c\u6362\u6d41\u3001\u7ba1\u9053\u64cd\u4f5c\u4ee5\u53ca\u80cc\u538b\u5904\u7406\u3002<\/p>\n1. Stream \u57fa\u7840<\/h2>\n
\n
fs.createReadStream<\/code>\uff09<\/li>\nfs.createWriteStream<\/code>\uff09<\/li>\nzlib.createGzip<\/code>\uff09<\/li>\nnet.Socket<\/code>\uff09<\/li>\n<\/ul>\n1.1 \u521b\u5efa\u7b80\u5355\u7684\u8bfb\u5199\u6d41<\/h3>\n
\nconst fs = require('fs');\n\n\/\/ \u521b\u5efa\u53ef\u8bfb\u6d41\nconst readableStream = fs.createReadStream('input.txt', {\n highWaterMark: 16 * 1024, \/\/ \u6bcf\u6b21\u8bfb\u53d6 16KB\uff0c\u9ed8\u8ba4 64KB\n encoding: 'utf8'\n});\n\n\/\/ \u521b\u5efa\u53ef\u5199\u6d41\nconst writableStream = fs.createWriteStream('output.txt');\n\n\/\/ \u76d1\u542c\u6570\u636e\u4e8b\u4ef6\nreadableStream.on('data', (chunk) => {\n console.log(`Received ${chunk.length} bytes of data.`);\n writableStream.write(chunk);\n});\n\nreadableStream.on('end', () => {\n console.log('Reading finished.');\n writableStream.end();\n});\n\n\/\/ \u9519\u8bef\u5904\u7406\nreadableStream.on('error', (err) => console.error('Read error:', err));\nwritableStream.on('error', (err) => console.error('Write error:', err));\n<\/code><\/pre>\nwritableStream.end()<\/code> \u4f1a\u5bfc\u81f4\u6587\u4ef6\u4e0d\u5b8c\u6574\u3002\u63a8\u8350\u4f7f\u7528 pipe<\/code> \u65b9\u6cd5\u3002<\/p>\n2. \u7ba1\u9053\uff08Pipe\uff09\u64cd\u4f5c<\/h2>\n
pipe<\/code> \u662f\u6d41\u6700\u5f3a\u5927\u7684\u529f\u80fd\uff0c\u5b83\u81ea\u52a8\u7ba1\u7406\u80cc\u538b\uff08backpressure\uff09\uff0c\u9632\u6b62\u6570\u636e\u79ef\u538b\u5bfc\u81f4\u5185\u5b58\u6ea2\u51fa\u3002<\/p>\n\nconst fs = require('fs');\n\nconst readable = fs.createReadStream('largefile.txt');\nconst writable = fs.createWriteStream('copy.txt');\n\nreadable.pipe(writable);\n\nwritable.on('finish', () => console.log('File copied successfully.'));\n<\/code><\/pre>\n2.1 \u94fe\u5f0f\u7ba1\u9053<\/h3>\n
\nconst fs = require('fs');\nconst zlib = require('zlib');\n\n\/\/ \u538b\u7f29\u6587\u4ef6\nfs.createReadStream('input.txt')\n .pipe(zlib.createGzip())\n .pipe(fs.createWriteStream('input.txt.gz'))\n .on('finish', () => console.log('Compression done.'));\n<\/code><\/pre>\nfinish<\/code> \u6216 close<\/code> \u4e8b\u4ef6\uff0c\u786e\u4fdd\u6240\u6709\u6570\u636e\u5df2\u5904\u7406\u5b8c\u6bd5\u3002<\/p>\n3. \u81ea\u5b9a\u4e49\u8f6c\u6362\u6d41<\/h2>\n
\nconst { Transform } = require('stream');\n\nclass UpperCaseTransform extends Transform {\n _transform(chunk, encoding, callback) {\n \/\/ chunk \u662f Buffer \u6216\u5b57\u7b26\u4e32\uff0c\u53d6\u51b3\u4e8e encoding\n this.push(chunk.toString().toUpperCase());\n callback();\n }\n}\n\n\/\/ \u4f7f\u7528\nconst upperCase = new UpperCaseTransform();\n\nprocess.stdin.pipe(upperCase).pipe(process.stdout);\n<\/code><\/pre>\n3.1 \u5b9e\u6218\uff1aCSV \u6570\u636e\u6e05\u6d17<\/h3>\n
\nconst fs = require('fs');\nconst { Transform } = require('stream');\n\n\/\/ \u521b\u5efa\u4e00\u4e2a\u8fc7\u6ee4\u8f6c\u6362\u6d41\nclass FilterTransform extends Transform {\n constructor(options) {\n super({ objectMode: true, ...options }); \/\/ \u4f7f\u7528\u5bf9\u8c61\u6a21\u5f0f\n this.header = null;\n }\n\n _transform(chunk, encoding, callback) {\n \/\/ \u5047\u8bbe chunk \u662f\u4e00\u884c\u5b57\u7b26\u4e32\n const line = chunk.toString().trim();\n if (!line) return callback();\n\n if (!this.header) {\n this.header = line.split(',');\n return callback();\n }\n\n const values = line.split(',');\n if (values[1] === 'valid') { \/\/ \u5047\u8bbe\u7b2c\u4e8c\u5217\u662f\u72b6\u6001\n const obj = {};\n this.header.forEach((key, i) => { obj[key] = values[i]; });\n this.push(JSON.stringify(obj) + '\\n');\n }\n callback();\n }\n}\n\n\/\/ \u4f7f\u7528\nconst readStream = fs.createReadStream('data.csv');\nconst writeStream = fs.createWriteStream('output.json');\n\nreadStream\n .pipe(new FilterTransform())\n .pipe(writeStream)\n .on('finish', () => console.log('CSV processed.'));\n<\/code><\/pre>\nobjectMode<\/code> \u540e\uff0c\u6d41\u5904\u7406\u7684\u662f\u5bf9\u8c61\u800c\u975e Buffer\uff0c\u4f46\u8981\u6ce8\u610f\u5185\u5b58\u7ba1\u7406\u3002<\/p>\n4. \u80cc\u538b\uff08Backpressure\uff09\u8be6\u89e3<\/h2>\n
pipe<\/code> \u65b9\u6cd5\u81ea\u52a8\u5904\u7406\u80cc\u538b\uff0c\u4f46\u624b\u52a8\u5b9e\u73b0\u65f6\u9700\u6ce8\u610f\u3002<\/p>\n4.1 \u624b\u52a8\u5904\u7406\u80cc\u538b<\/h3>\n
\nconst fs = require('fs');\n\nconst readable = fs.createReadStream('large.txt');\nconst writable = fs.createWriteStream('copy.txt');\n\nreadable.on('data', (chunk) => {\n const canContinue = writable.write(chunk);\n if (!canContinue) {\n console.log('Backpressure detected, pausing read.');\n readable.pause();\n writable.once('drain', () => {\n console.log('Drain event, resuming read.');\n readable.resume();\n });\n }\n});\n\nreadable.on('end', () => writable.end());\n<\/code><\/pre>\npipe<\/code> \u6216 pipeline<\/code>\uff08Node.js 10+\uff09\u6765\u5904\u7406\u80cc\u538b\u3002<\/p>\n5. \u4f7f\u7528 pipeline \u66ff\u4ee3 pipe<\/h2>\n
pipeline<\/code> \u662f Node.js 10 \u5f15\u5165\u7684\u6a21\u5757\u5316\u65b9\u6cd5\uff0c\u5b83\u81ea\u52a8\u5904\u7406\u9519\u8bef\u548c\u6e05\u7406\u3002<\/p>\n\nconst { pipeline } = require('stream');\nconst fs = require('fs');\nconst zlib = require('zlib');\n\npipeline(\n fs.createReadStream('input.txt'),\n zlib.createGzip(),\n fs.createWriteStream('input.txt.gz'),\n (err) => {\n if (err) {\n console.error('Pipeline failed:', err);\n } else {\n console.log('Pipeline succeeded.');\n }\n }\n);\n<\/code><\/pre>\npipeline<\/code> \u5728\u53d1\u751f\u9519\u8bef\u65f6\u4f1a\u81ea\u52a8\u9500\u6bc1\u6240\u6709\u6d41\uff0c\u907f\u514d\u8d44\u6e90\u6cc4\u6f0f\u3002<\/p>\n6. \u5b9e\u6218\uff1a\u5927\u6587\u4ef6\u884c\u5904\u7406<\/h2>\n
readline<\/code> \u6a21\u5757\u3002<\/p>\n\nconst fs = require('fs');\nconst readline = require('readline');\n\nasync function processLargeFile(filePath) {\n const fileStream = fs.createReadStream(filePath);\n\n const rl = readline.createInterface({\n input: fileStream,\n crlfDelay: Infinity\n });\n\n let lineCount = 0;\n for await (const line of rl) {\n \/\/ \u5904\u7406\u6bcf\u4e00\u884c\n lineCount++;\n if (lineCount % 100000 === 0) {\n console.log(`Processed ${lineCount} lines`);\n }\n }\n console.log(`Total lines: ${lineCount}`);\n}\n\nprocessLargeFile('huge.log').catch(console.error);\n<\/code><\/pre>\nreadline<\/code> \u5185\u90e8\u4f7f\u7528\u4e86\u6d41\uff0c\u4f46\u5b83\u662f\u6309\u884c\u5206\u5272\u7684\uff0c\u9002\u5408\u5904\u7406\u6587\u672c\u6587\u4ef6\u3002<\/p>\n7. \u5e38\u89c1\u9677\u9631\u4e0e\u6700\u4f73\u5b9e\u8df5<\/h2>\n
7.1 \u4e0d\u8981\u5fd8\u8bb0\u9519\u8bef\u5904\u7406<\/h3>\n
\nconst fs = require('fs');\nconst stream = fs.createReadStream('nonexistent.txt');\nstream.on('error', (err) => console.error('Error:', err.message));\n<\/code><\/pre>\n7.2 \u907f\u514d\u5728 ‘data’ \u4e8b\u4ef6\u4e2d\u6267\u884c\u540c\u6b65\u64cd\u4f5c<\/h3>\n
7.3 \u4f7f\u7528 highWaterMark \u63a7\u5236\u7f13\u51b2\u533a\u5927\u5c0f<\/h3>\n
highWaterMark<\/code>\u3002<\/p>\n7.4 \u6d41\u4e0d\u662f\u4e07\u80fd\u7684<\/h3>\n
fs.readFile<\/code> \u53ef\u80fd\u66f4\u7b80\u5355\u3002<\/p>\n\u603b\u7ed3<\/h2>\n
pipe<\/code> \u6216 pipeline<\/code>\uff0c\u4f60\u53ef\u4ee5\u8f7b\u677e\u7ec4\u5408\u591a\u4e2a\u6d41\uff0c\u5b9e\u73b0\u9ad8\u6548\u7684\u6570\u636e\u5904\u7406\u3002\u672c\u6587\u4ecb\u7ecd\u4e86\u6d41\u7684\u57fa\u672c\u7528\u6cd5\u3001\u81ea\u5b9a\u4e49\u8f6c\u6362\u6d41\u3001\u80cc\u538b\u5904\u7406\u4ee5\u53ca\u5b9e\u6218\u6848\u4f8b\u3002\u638c\u63e1\u8fd9\u4e9b\u6280\u5de7\u540e\uff0c\u4f60\u5c06\u80fd\u591f\u4f18\u96c5\u5730\u5904\u7406 GB \u7ea7\u522b\u7684\u6570\u636e\u3002<\/p>\n\n
stream.finished()<\/code> \u548c stream.destroy()<\/code> \u8fdb\u884c\u8d44\u6e90\u6e05\u7406<\/li>\nthrough2<\/code>\u3001split2<\/code> \u7b49<\/li>\n<\/ul>\n","protected":false},"excerpt":{"rendered":"