blocks|key|422923|text|drain背后的想法是，您可以在这里使用它进行测试：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|422924|var+fs+=+require('fs');
var+stream+=+fs.createWriteStream('someFile.txt',+{flags:+'w'});

var+lines;
while+(lines+=+getLines())+{
++++for+(var+i+=+0;+i+<+lines.length;+i%2B%2B)+{
++++++++stream.write(lines[i]);+//<--+the+place+to+test
++++}
}|code-block|syntax|javascript|422925|但你不是。因此，您需要重新设计以使其成为“可重入的”。|422926|var+fs+=+require('fs');
var+stream+=+fs.createWriteStream('someFile.txt',+{flags:+'w'});

var+lines;
while+(lines+=+getLines())+{
++++for+(var+i+=+0;+i+<+lines.length;+i%2B%2B)+{
++++++++var+written+=+stream.write(lines[i]);+//<--+the+place+to+test
++++++++if+(!written){
+++++++++++//do+something+here+to+wait+till+you+can+safely+write+again
+++++++++++//this+means+prepare+a+buffer+and+wait+till+you+can+come+back+to+finish
+++++++++++//++lines[i]+->+remainder
++++++++}
++++}
}|422927|然而，这是否意味着您在等待时也需要继续缓冲getLines？|422928|var+fs+=+require('fs');
var+stream+=+fs.createWriteStream('someFile.txt',+{flags:+'w'});

var+lines,
++++buffer+=+{
+++++remainingLines+=+[]
++++};
while+(lines+=+getLines())+{
++++for+(var+i+=+0;+i+<+lines.length;+i%2B%2B)+{
++++++++var+written+=+stream.write(lines[i]);+//<--+the+place+to+test
++++++++if+(!written){
+++++++++++//do+something+here+to+wait+till+you+can+safely+write+again
+++++++++++//this+means+prepare+a+buffer+and+wait+till+you+can+come+back+to+finish
+++++++++++//++lines[i]+->+remainder
+++++++++++buffer.remainingLines+=+lines.slice(i);
+++++++++++break;
+++++++++++//notice+there's+no+way+to+re-run+this+once+we+leave+here.
++++++++}
++++}
}

stream.on('drain',function(){
++if+(buffer.remainingLines.length){
++++for+(var+i+=+0;+i+<+buffer.remainingLines.length;+i%2B%2B)+{
++++++var+written+=+stream.write(buffer.remainingLines[i]);+//<--+the+place+to+test
++++++if+(!written){
+++++++//do+something+here+to+wait+till+you+can+safely+write+again
+++++++//this+means+prepare+a+buffer+and+wait+till+you+can+come+back+to+finish
+++++++//++lines[i]+->+remainder
+++++++buffer.remainingLines+=+lines.slice(i);
++++++}
++++}
++}
});|422929|entityMap^0|0|0|0|0|0|0^^$0|@$1|2|3|4|5|6|7|Q|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|R|8|@]|9|@]|A|$E|F]]|$1|G|3|H|5|6|7|S|8|@]|9|@]|A|$]]|$1|I|3|J|5|D|7|T|8|@]|9|@]|A|$E|F]]|$1|K|3|L|5|6|7|U|8|@]|9|@]|A|$]]|$1|M|3|N|5|D|7|V|8|@]|9|@]|A|$E|F]]|$1|O|3|-4|5|6|7|W|8|@]|9|@]|A|$]]]|P|$]]

The idea behind drain is that you would use it to test here:

<pre><code>var fs = require('fs');
var stream = fs.createWriteStream('someFile.txt', {flags: 'w'});

var lines;
while (lines = getLines()) {
 for (var i = 0; i &lt; lines.length; i++) {
 stream.write(lines[i]); //&lt;-- the place to test
 }
}
</code></pre>

which you're not. So you would need to rearchitect to make it "reentrant".

<pre><code>var fs = require('fs');
var stream = fs.createWriteStream('someFile.txt', {flags: 'w'});

var lines;
while (lines = getLines()) {
 for (var i = 0; i &lt; lines.length; i++) {
 var written = stream.write(lines[i]); //&lt;-- the place to test
 if (!written){
 //do something here to wait till you can safely write again
 //this means prepare a buffer and wait till you can come back to finish
 // lines[i] -&gt; remainder
 }
 }
}
</code></pre>

However, does this mean that you need to keep buffering getLines as well while you wait?

<pre><code>var fs = require('fs');
var stream = fs.createWriteStream('someFile.txt', {flags: 'w'});

var lines,
 buffer = {
 remainingLines = []
 };
while (lines = getLines()) {
 for (var i = 0; i &lt; lines.length; i++) {
 var written = stream.write(lines[i]); //&lt;-- the place to test
 if (!written){
 //do something here to wait till you can safely write again
 //this means prepare a buffer and wait till you can come back to finish
 // lines[i] -&gt; remainder
 buffer.remainingLines = lines.slice(i);
 break;
 //notice there's no way to re-run this once we leave here.
 }
 }
}

stream.on('drain',function(){
 if (buffer.remainingLines.length){
 for (var i = 0; i &lt; buffer.remainingLines.length; i++) {
 var written = stream.write(buffer.remainingLines[i]); //&lt;-- the place to test
 if (!written){
 //do something here to wait till you can safely write again
 //this means prepare a buffer and wait till you can come back to finish
 // lines[i] -&gt; remainder
 buffer.remainingLines = lines.slice(i);
 }
 }
 }
});
</code></pre>

blocks|key|637499|text|这就是我最终如何做到的。其背后的思想是创建实现ReadStream接口的可读流，然后使用pipe()方法将数据管道到可写流。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|637500|var+fs+=+require('fs');
var+writeStream+=+fs.createWriteStream('someFile.txt',+{+flags+:+'w'+});
var+readStream+=+new+MyReadStream();

readStream.pipe(writeStream);
writeStream.on('close',+function+()+{
++++console.log('All+done!');
});|code-block|syntax|javascript|637501|可以从mongoose+QueryStream中获取MyReadStream类的示例。|637502|entityMap|0|LINK|mutability|MUTABLE|url|http://nodejs.org/api/stream.html#stream_readable_stream|1|https://github.com/LearnBoost/mongoose/blob/master/lib/querystream.js^0|18|6|N|A|0|0|0|Q|C|C|B|1|0^^$0|@$1|2|3|4|5|6|7|W|8|@$9|X|A|Y|B|C]]|D|@$9|Z|A|10|1|11]]|E|$]]|$1|F|3|G|5|H|7|12|8|@]|D|@]|E|$I|J]]|$1|K|3|L|5|6|7|13|8|@$9|14|A|15|B|C]]|D|@$9|16|A|17|1|18]]|E|$]]|$1|M|3|-4|5|6|7|19|8|@]|D|@]|E|$]]]|N|$O|$5|P|Q|R|E|$S|T]]|U|$5|P|Q|R|E|$S|V]]]]

That's how I finally did it. The idea behind is to create readable stream implementing <a href="http://nodejs.org/api/stream.html#stream_readable_stream">ReadStream</a> interface and then use <code>pipe()</code> method to pipe data to writable stream.

<pre><code>var fs = require('fs');
var writeStream = fs.createWriteStream('someFile.txt', { flags : 'w' });
var readStream = new MyReadStream();

readStream.pipe(writeStream);
writeStream.on('close', function () {
 console.log('All done!');
});
</code></pre>

The example of <code>MyReadStream</code> class can be taken from mongoose <a href="https://github.com/LearnBoost/mongoose/blob/master/lib/querystream.js">QueryStream</a>.

blocks|key|637573|text|我发现在处理大文件时，流是一种性能不佳的方法--这是因为你不能设置足够的输入缓冲区大小(至少我不知道有一个好的方法可以做到这一点)。这就是我要做的：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|637574|var+fs+=+require('fs');

var+i+=+fs.openSync('input.txt',+'r');
var+o+=+fs.openSync('output.txt',+'w');

var+buf+=+new+Buffer(1024+*+1024),+len,+prev+=+'';

while(len+=+fs.readSync(i,+buf,+0,+buf.length))+{

++++var+a+=+(prev+%2B+buf.toString('ascii',+0,+len)).split('\n');
++++prev+=+len+===+buf.length+?+'\n'+%2B+a.splice(a.length+-+1)[0]+:+'';

++++var+out+=+'';
++++a.forEach(function(line)+{

++++++++if(!line)
++++++++++++return;

++++++++//+do+something+with+your+line+here

++++++++out+%2B=+line+%2B+'\n';
++++});

++++var+bout+=+new+Buffer(out,+'ascii');
++++fs.writeSync(o,+bout,+0,+bout.length);
}

fs.closeSync(o);
fs.closeSync(i);|code-block|syntax|javascript|637575|entityMap^0|0|0^^$0|@$1|2|3|4|5|6|7|I|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|J|8|@]|9|@]|A|$E|F]]|$1|G|3|-4|5|6|7|K|8|@]|9|@]|A|$]]]|H|$]]

I found streams to be a poor performing way to deal with large files - this is because you cannot set an adequate input buffer size (at least I'm not aware of a good way to do it). This is what I do:

<pre><code>var fs = require('fs');

var i = fs.openSync('input.txt', 'r');
var o = fs.openSync('output.txt', 'w');

var buf = new Buffer(1024 * 1024), len, prev = '';

while(len = fs.readSync(i, buf, 0, buf.length)) {

 var a = (prev + buf.toString('ascii', 0, len)).split('\n');
 prev = len === buf.length ? '\n' + a.splice(a.length - 1)[0] : '';

 var out = '';
 a.forEach(function(line) {

 if(!line)
 return;

 // do something with your line here

 out += line + '\n';
 });

 var bout = new Buffer(out, 'ascii');
 fs.writeSync(o, bout, 0, bout.length);
}

fs.closeSync(o);
fs.closeSync(i);
</code></pre>

blocks|key|423077|text|处理这个问题的最干净的方法是让你的行生成器成为一个readable+stream+--让我们称之为lineReader。然后，下面的代码将自动处理缓冲区并很好地为您排出：|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|423078|lineReader.pipe(fs.createWriteStream('someFile.txt'));|code-block|syntax|javascript|423079|如果你不想创建一个可读的流，你可以监听write的输出，看看是否有足够的缓冲区，并像这样响应：|423080|var+i+=+0,+n+=+lines.length;
function+write+()+{
++if+(i+===+n)+return;++//+A+callback+could+go+here+to+know+when+it's+done.
++while+(stream.write(lines[i%2B%2B])+&&+i+<+n);
++stream.once('drain',+write);
}
write();++//+Initial+call.|423081|这种情况的一个更长的例子可以在here中找到。|423082|entityMap|0|LINK|mutability|MUTABLE|url|http://nodejs.org/api/stream.html#stream_class_stream_readable_1|1|http://nodejs.org/api/stream.html#stream_event_drain^0|1D|A|P|F|0|0|0|J|5|0|0|F|4|1|0^^$0|@$1|2|3|4|5|6|7|10|8|@$9|11|A|12|B|C]]|D|@$9|13|A|14|1|15]]|E|$]]|$1|F|3|G|5|H|7|16|8|@]|D|@]|E|$I|J]]|$1|K|3|L|5|6|7|17|8|@$9|18|A|19|B|C]]|D|@]|E|$]]|$1|M|3|N|5|H|7|1A|8|@]|D|@]|E|$I|J]]|$1|O|3|P|5|6|7|1B|8|@]|D|@$9|1C|A|1D|1|1E]]|E|$]]|$1|Q|3|-4|5|6|7|1F|8|@]|D|@]|E|$]]]|R|$S|$5|T|U|V|E|$W|X]]|Y|$5|T|U|V|E|$W|Z]]]]

The cleanest way to handle this is to make your line generator a <a href="http://nodejs.org/api/stream.html#stream_class_stream_readable_1" rel="noreferrer">readable stream</a> - let's call it <code>lineReader</code>. Then the following would automatically handle the buffers and draining nicely for you:

<pre><code>lineReader.pipe(fs.createWriteStream('someFile.txt'));
</code></pre>

If you don't want to make a readable stream, you can listen to <code>write</code>'s output for buffer-fullness and respond like this:

<pre><code>var i = 0, n = lines.length;
function write () {
 if (i === n) return; // A callback could go here to know when it's done.
 while (stream.write(lines[i++]) &amp;&amp; i &lt; n);
 stream.once('drain', write);
}
write(); // Initial call.
</code></pre>

A longer example of this situation can be found <a href="http://nodejs.org/api/stream.html#stream_event_drain" rel="noreferrer">here</a>.

blocks|key|423123|text|对这个问题的几个建议答案完全没有抓住关于流的要点。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|423124|此模块可以帮助https://www.npmjs.org/package/JSONStream|offset|length|423125|但是，让我们假设所描述的情况，并自己编写代码。您将以流的形式从MongoDB读取数据，默认情况下ObjectMode+=+true。|423126|这将导致问题，如果您尝试直接流到文件-类似于“无效的非字符串/缓冲区块”错误。|423127|这类问题的解决方案非常简单。|423128|只需将另一个转换放在可读和可写之间，以便将可读对象适当地适配为可写字符串。|423129|示例代码解决方案：|423130|var+fs+=+require('fs'),
++++writeStream+=+fs.createWriteStream('./out'+%2B+process.pid,+{flags:+'w',+encoding:+'utf-8'+}),
++++stream+=+require('stream'),
++++stringifier+=+new+stream.Transform();
stringifier._writableState.objectMode+=+true;
stringifier._transform+=+function+(data,+encoding,+done)+{
++++this.push(JSON.stringify(data));
++++this.push('\n');
++++done();
}
rowFeedDao.getRowFeedsStream(merchantId,+jobId)
.pipe(stringifier)
.pipe(writeStream).on('error',+function+(err)+{
+++//+handle+error+condition
}|code-block|syntax|javascript|423131|entityMap|0|LINK|mutability|MUTABLE|url|https://www.npmjs.org/package/JSONStream^0|0|7|14|0|0|0|0|0|0|0|0^^$0|@$1|2|3|4|5|6|7|12|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|13|8|@]|9|@$D|14|E|15|1|16]]|A|$]]|$1|F|3|G|5|6|7|17|8|@]|9|@]|A|$]]|$1|H|3|I|5|6|7|18|8|@]|9|@]|A|$]]|$1|J|3|K|5|6|7|19|8|@]|9|@]|A|$]]|$1|L|3|M|5|6|7|1A|8|@]|9|@]|A|$]]|$1|N|3|O|5|6|7|1B|8|@]|9|@]|A|$]]|$1|P|3|Q|5|R|7|1C|8|@]|9|@]|A|$S|T]]|$1|U|3|-4|5|6|7|1D|8|@]|9|@]|A|$]]]|V|$W|$5|X|Y|Z|A|$10|11]]]]

Several suggested answers to this question have missed the point about streams altogether.

This module can help <a href="https://www.npmjs.org/package/JSONStream" rel="nofollow">https://www.npmjs.org/package/JSONStream</a>

However, lets suppose the situation as described and write the code ourselves. You are reading from a MongoDB as a stream, with ObjectMode = true by default.

This will lead to issues if you try to directly stream to file - something like "Invalid non-string/buffer chunk" error.

The solution to this type of problem is very simple.

Just put another Transform in between the readable and writeable to adapt the Object readable to a String writeable appropriately. 

Sample Code Solution:

<pre><code>var fs = require('fs'),
 writeStream = fs.createWriteStream('./out' + process.pid, {flags: 'w', encoding: 'utf-8' }),
 stream = require('stream'),
 stringifier = new stream.Transform();
stringifier._writableState.objectMode = true;
stringifier._transform = function (data, encoding, done) {
 this.push(JSON.stringify(data));
 this.push('\n');
 done();
}
rowFeedDao.getRowFeedsStream(merchantId, jobId)
.pipe(stringifier)
.pipe(writeStream).on('error', function (err) {
 // handle error condition
}
</code></pre>

blocks|key|637708|text|如果您碰巧没有输入流，就不能轻松地使用管道。上面的方法对我都不起作用，排出事件不会触发。解决方法如下(基于Tylers答案)：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|637709|var+lines[];+//+some+very+large+array
var+i+=+0;

function+write()+{
++++if+(i+<+lines.length)++{
++++++++wstream.write(lines[i]),+function(err){
++++++++++++if+(err)+{
++++++++++++++++console.log(err);
++++++++++++}+else+{
++++++++++++++++i%2B%2B;
++++++++++++++++write();
++++++++++++}
++++++++});
++++}+else+{
++++++++wstream.end();
++++++++console.log("done");
++++}
};
write();|code-block|syntax|javascript|637710|entityMap^0|0|0^^$0|@$1|2|3|4|5|6|7|I|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|J|8|@]|9|@]|A|$E|F]]|$1|G|3|-4|5|6|7|K|8|@]|9|@]|A|$]]]|H|$]]

If you do not happen to have an input stream you cannot easily use pipe.
None of the above worked for me, the drain event doesn't fire. Solved as follows (based on Tylers answer):

<pre><code>var lines[]; // some very large array
var i = 0;

function write() {
 if (i &lt; lines.length) {
 wstream.write(lines[i]), function(err){
 if (err) {
 console.log(err);
 } else {
 i++;
 write();
 }
 });
 } else {
 wstream.end();
 console.log("done");
 }
};
write();
</code></pre>

I'm writing a large file with node.js using a <a href="http://nodejs.org/docs/latest/api/streams.html#writable_Stream" rel="noreferrer">writable stream</a>:

<pre><code>var fs = require('fs');
var stream = fs.createWriteStream('someFile.txt', { flags : 'w' });

var lines;
while (lines = getLines()) {
 for (var i = 0; i &lt; lines.length; i++) {
 stream.write( lines[i] );
 }
}
</code></pre>

I'm wondering if this scheme is safe without using <a href="http://nodejs.org/docs/latest/api/streams.html#event_drain_" rel="noreferrer"><code>drain</code></a> event? If it is not (which I think is the case), what is the pattern for writing an arbitrary large data to a file?

Writing large files with Node.js

数据管道

MongoDB

JSON

我正在使用用node.js编写一个大文件var fs     = require('fs');var stream = fs.createWriteStream('someFile.txt', { flags : 'w' });var lines;while (lines = getLines()) {    for ...

问使用Node.js编写大文件
EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问使用Node.js编写大文件EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问使用Node.js编写大文件
EN