我是node.js的新手。问题是从2个(或更多)已排序数据的流中读取数据,并产生它们的“已排序合并”。例如:
Stream A: 1 5 6 8
Stream B: 2 3 4 7
========================
Result: 1 2 3 4 5 6 7 8
在C ++ / Java / C#中,这有一个非常明显的解决方案,类似于:
BufferedReader[] readers = new BufferedReader[2];
String[] lines = new String[2];
// fill lines with initial values from both readers
// ...
while (true) {
int earliestIndex = -1;
// ....
// determine earliestIndex by looping over lines and comparing them
if (earliestIndex < 0) break;
String line = lines[earliestIndex];
// do something with line
System.out.println(line);
// advance reader
lines[earliestIndex] = readers[earliestIndex].readLine();
}
但是在节点中,这似乎相当困难。有任何想法吗?
这是我最终想出的解决方案。我正在使用node-line-reader逐行读取流(文件流,但是可以轻松更改):
var LineReader = require('node-line-reader').LineReader;
var files = ['c:\\temp\\1.txt', 'c:\\temp\\2.txt'];
var readers = [];
var lines = [];
var readWhile = function (done) {
var earliestIndex = -1;
var earliest = MAX_VALUE;
for (i = 0; i < lines.length; i++) {
var l = lines[i];
var value = parseInt(l);
if (value < earliest) {
earliest = value;
earliestIndex = i;
}
}
if (earliestIndex < 0) {
done();
return;
}
var line = lines[earliestIndex];
console.log('Read from ' + files[earliestIndex] + ': ' + line);
readers[earliestIndex].nextLine(function (err, line) {
if (err) throw err;
lines[earliestIndex] = line;
process.nextTick(function () {
readWhile(done);
});
});
}
new Promise(function (success, error) {
for (i = 0; i < files.length; i++) {
var reader = new LineReader(files[i]);
readers.push(reader);
new Promise(function (success, failure) {
reader.nextLine(function (err, line) {
if (err) failure(err);
lines.push(line);
success();
});
}).then(function (data) {
if (lines.length == files.length) success();
});
}
}).then(function (data) {
return new Promise(function (success, failure) {
readWhile(success);
});
}).then(function() {
console.log('All done');
}, function (err) {
console.log('Error: ' + err);
});
我只是有同样的问题。我最终使用了https://www.npmjs.com/package/sorted-union-stream
仅用一行就解决了问题。