Nodejs - 如何将函数应用于两个流的每一行组合？

debugcn 发表于 Dev

血腥

我正在流式传输两个无法放入内存的大文件，我需要逐行比较这两个文件并应用需要积累一些信息的比较函数。

假设我有两个流：const stream1, stream2. 还有一些我需要积累的结果。例如，假设它是两个文件中相同位置相同行数的计数：let count.

这是我尝试的方法：

const stream1 = getLineByLineReadable1Somehow(), stream2 = getLineByLineReadable2Somehow();

let count = 0;

stream1.on('readable', () => {
    stream2.on('readble', () => {
        let line1, line2;
        while (line1 = stream1.read()) {
            line2 = stream2.read();
            if (line1 !== line2) count++;
        }
    });
});

console.log(count);

只要这两个文件很小，这就可以正常工作，但是当文件变大时就完全崩溃了。最后提供的计数不正确。不确定发生了什么，但我认为这与readable两个流之一被多次触发的事件有关。在某些时候，两个流中的一个会暂时吐出一堆null。

感谢您对此事的任何帮助。谢谢！

标准数据库--

问题是在大文件中，新行到达的事件可能非常不规则。所以有必要有一个中间缓冲区。这是一个工作片段：

var readline = require('readline');
var fs = require('fs');

var rs1 = fs.createReadStream('1.log');
var r1 = readline.createInterface({
    input: rs1
});

var rs2 = fs.createReadStream('2.log');
var r2 = readline.createInterface({
    input: rs2
});

var stats = {
    count: [0, 0],
    common: 0,
    closed: 0,
    lines: [
        [],
        []
    ]
}

function compare(line, i) {
    stats.count[i]++;
    if (stats.closed === 0 || stats.lines[1 - i].length > 0) stats.lines[i].push(line);
    while (stats.lines[0].length > 0 && stats.lines[1].length > 0) {
        var L1 = stats.lines[0].shift();
        var L2 = stats.lines[1].shift();
        if (L1 === L2) stats.common++;
    }
}

function close() {
    if (++stats.closed === 2) console.log(stats.count, stats.common, stats.lines[0].length, stats.lines[1].length);
}

r1.on('line', (line) => compare(line, 0))
r2.on('line', (line) => compare(line, 1))

r1.on('close', close);
r2.on('close', close);

本文收集自互联网，转载请注明来源。

如有侵权，请联系[email protected] 删除。