在大量阅读文档和源代码之后，进行了大量的反复试验和一些测试。我已经为我的问题想出了一个解决方案。我可以复制并粘贴我的解决方案，但为了完整起见，我将在这里解释我的发现。
用管道处理背压由几个部分组成。我们有将数据写入Writable 的Readable。 Readable 为Writable 提供回调，它可以告诉Readable 它已准备好接收新的数据块。阅读部分比较简单。 Readable 有一个内部缓冲区。使用Readable.push() 会将数据添加到缓冲区。当数据被读取时，它将来自这个内部缓冲区。接下来我们可以使用Readable.readableHighWaterMark 和Readable.readableLength 来确保我们不会一次推送太多数据。
Readable.readableHighWaterMark - Readable.readableLength
是我们应该推送到这个内部缓冲区的最大字节数。
所以这意味着，由于我们想同时读取两个Readable 流，我们需要两个Writable 流来控制流。要合并数据，我们需要自己缓冲它，因为（据我所知）Writable 流中没有内部缓冲区。所以双工流将是最好的选择，因为我们想要处理缓冲、写入和读取我们自己。
写作
现在让我们开始编写代码。为了控制多个流的状态，我们将创建一个状态接口。如下所示：
declare type StreamCallback = (error?: Error | null) => void;
interface MergingState {
callback: StreamCallback;
queue: BufferList;
highWaterMark: number;
size: number;
finalizing: boolean;
}
callback 保存由 write 或 final 提供的最后一个回调（稍后我们将讨论 final）。 highWaterMark 表示我们的queue 的最大大小，size 是我们当前队列的大小。最后finalizing 标志表示当前队列是最后一个队列。因此，一旦队列为空，我们就完成了对属于该状态的流的读取。
BufferList 是用于构建流的internal Nodejs implementation 的副本。
如前所述，可写对象处理背压，因此我们的两个可写对象的通用方法如下所示：
/**
* Method to write to provided state if it can
*
* (Will unshift the bytes that cannot be written back to the source)
*
* @param src the readable source that writes the chunk
* @param chunk the chunk to be written
* @param encoding the chunk encoding, currently not used
* @param cb the streamCallback provided by the writing state
* @param state the state which should be written to
*/
private writeState(src: Readable, chunk: Buffer, encoding: string, cb: StreamCallback, state: MergingState): void {
this.mergeNextTick();
const bytesAvailable = state.highWaterMark - state.size;
if (chunk.length <= bytesAvailable) {
// save to write to our local buffer
state.queue.push(chunk);
state.size += chunk.length;
if (chunk.length === bytesAvailable) {
// our queue is full, so store our callback
this.stateCallbackAndSet(state, cb);
} else {
// we still have some space, so we can call the callback immediately
cb();
}
return;
}
if (bytesAvailable === 0) {
// no space available unshift entire chunk
src.unshift(chunk);
} else {
state.size += bytesAvailable;
const leftOver = Buffer.alloc(chunk.length - bytesAvailable);
chunk.copy(leftOver, 0, bytesAvailable);
// push amount of bytes available
state.queue.push(chunk.slice(0, bytesAvailable));
// unshift what we cannot fit in our queue
src.unshift(leftOver);
}
this.stateCallbackAndSet(state, cb);
}
首先我们检查有多少空间可用于缓冲。如果我们的整个块有足够的空间，我们将缓冲它。如果没有可用空间，我们会将缓冲区移至其可读源。如果有一些可用空间，我们只会取消我们无法容纳的空间。如果我们的缓冲区已满，我们将存储请求新块的回调。如果有空间，我们将请求下一个块。
this.mergeNextTick() 被调用是因为我们的状态已经改变，并且应该在下一个滴答声中读取：
private mergeNextTick(): void {
if (!this.mergeSync) {
// make sure it is only called once per tick
// we don't want to call it multiple times
// since there will be nothing left to read the second time
this.mergeSync = true;
process.nextTick(() => this._read(this.readableHighWaterMark));
}
}
this.stateCallbackAndSet 是一个辅助函数，它只会调用我们最后的回调，以确保我们不会进入使我们的流停止流动的状态。并且会提供新的。
/**
* Helper function to call the callback if it exists and set the new callback
* @param state the state which holds the callback
* @param cb the new callback to be set
*/
private stateCallbackAndSet(state: MergingState, cb: StreamCallback): void {
if (!state) {
return;
}
if (state.callback) {
const callback = state.callback;
// do callback next tick, such that we can't get stuck in a writing loop
process.nextTick(() => callback());
}
state.callback = cb;
}
阅读
现在在阅读方面，这是我们处理选择正确流的部分。
首先我们的函数读取状态，这非常简单。它读取它能够读取的字节数。它返回写入的字节数，这对我们的其他函数很有用。
/**
* Method to read the provided state if it can
*
* @param size the number of bytes to consume
* @param state the state from which needs to be read
* @returns the amount of bytes read
*/
private readState(size: number, state: MergingState): number {
if (state.size === 0) {
// our queue is empty so we read 0 bytes
return 0;
}
let buffer = null;
if (state.size < size) {
buffer = state.queue.consume(state.size, false);
} else {
buffer = state.queue.consume(size, false);
}
this.push(buffer);
this.stateCallbackAndSet(state, null);
state.size -= buffer.length;
return buffer.length;
}
doRead 方法是所有合并发生的地方：它获取 nextMergingIndex。如果合并索引是END，那么我们可以读取writingState，直到流结束。如果我们在合并索引处，我们从mergingState 读取。否则，我们会从writingState 中读取尽可能多的内容，直到我们到达下一个合并索引。
/**
* Method to read from the correct Queue
*
* The doRead method is called multiple times by the _read method until
* it is satisfied with the returned size, or until no more bytes can be read
*
* @param n the number of bytes that can be read until highWaterMark is hit
* @throws Errors when something goes wrong, so wrap this method in a try catch.
* @returns the number of bytes read from either buffer
*/
private doRead(n: number): number {
// first check all constants below 0,
// which is only Merge.END right now
const nextMergingIndex = this.getNextMergingIndex();
if (nextMergingIndex === Merge.END) {
// read writing state until the end
return this.readWritingState(n);
}
const bytesToNextIndex = nextMergingIndex - this.index;
if (bytesToNextIndex === 0) {
// We are at the merging index, thus should read merging queue
return this.readState(n, this.mergingState);
}
if (n <= bytesToNextIndex) {
// We are safe to read n bytes
return this.readWritingState(n);
}
// read the bytes until the next merging index
return this.readWritingState(bytesToNextIndex);
}
readWritingState读取状态并更新索引：
/**
* Method to read from the writing state
*
* @param n maximum number of bytes to be read
* @returns number of bytes written.
*/
private readWritingState(n: number): number {
const bytesWritten = this.readState(n, this.writingState);
this.index += bytesWritten;
return bytesWritten;
}
合并
为了选择要合并的流，我们将使用生成器函数。生成器函数产生一个索引和一个要在该索引处合并的流：
export interface MergingStream { index: number; stream: Readable; }
在doRead getNextMergingIndex() 被调用。该函数返回下一个MergingStream 的索引。如果没有下一个 mergingStream，则调用生成器来获取新的 mergingStream。如果没有新的合并流，我们将返回END。
/**
* Method to get the next merging index.
*
* Also fetches the next merging stream if merging stream is null
*
* @returns the next merging index, or Merge.END if there is no new mergingStream
* @throws Error when invalid MergingStream is returned by streamGenerator
*/
private getNextMergingIndex(): number {
if (!this.mergingStream) {
this.setNewMergeStream(this.streamGenerator.next().value);
if (!this.mergingStream) {
return Merge.END;
}
}
return this.mergingStream.index;
}
在setNewMergeStream 中，我们正在创建一个新的Writable，我们可以将新的合并流通过管道传输到其中。对于我们的Writable，我们需要处理写入回调以写入我们的状态，并需要处理最终回调以处理最后一个块。我们也不应该忘记重置我们的状态。
/**
* Method to set the new merging stream
*
* @throws Error when mergingStream has an index less than the current index
*/
private setNewMergeStream(mergingStream?: MergingStream): void {
if (this.mergingStream) {
throw new Error('There already is a merging stream');
}
// Set a new merging stream
this.mergingStream = mergingStream;
if (mergingStream == null || mergingStream.index === Merge.END) {
// set new state
this.mergingState = newMergingState(this.writableHighWaterMark);
// We're done, for now...
// mergingStream will be handled further once nextMainStream() is called
return;
}
if (mergingStream.index < this.index) {
throw new Error('Cannot merge at ' + mergingStream.index + ' because current index is ' + this.index);
}
// Create a new writable our new mergingStream can write to
this.mergeWriteStream = new Writable({
// Create a write callback for our new mergingStream
write: (chunk, encoding, cb) => this.writeMerge(mergingStream.stream, chunk, encoding, cb),
final: (cb: StreamCallback) => {
this.onMergeEnd(mergingStream.stream, cb);
},
});
// Create a new mergingState for our new merging stream
this.mergingState = newMergingState(this.mergeWriteStream.writableHighWaterMark);
// Pipe our new merging stream to our sink
mergingStream.stream.pipe(this.mergeWriteStream);
}
完成
该过程的最后一步是处理我们的最终块。这样我们就知道何时结束合并并可以发送结束块。在我们的主读取循环中，我们首先读取直到我们的doRead() 方法连续两次返回0，或者已经填满了我们的读取缓冲区。一旦发生这种情况，我们将结束我们的读取循环并检查我们的状态以查看它们是否已完成。
public _read(size: number): void {
if (this.finished) {
// we've finished, there is nothing to left to read
return;
}
this.mergeSync = false;
let bytesRead = 0;
do {
const availableSpace = this.readableHighWaterMark - this.readableLength;
bytesRead = 0;
READ_LOOP: while (bytesRead < availableSpace && !this.finished) {
try {
const result = this.doRead(availableSpace - bytesRead);
if (result === 0) {
// either there is nothing in our buffers
// or our states are outdated (since they get updated in doRead)
break READ_LOOP;
}
bytesRead += result;
} catch (error) {
this.emit('error', error);
this.push(null);
this.finished = true;
}
}
} while (bytesRead > 0 && !this.finished);
this.handleFinished();
}
然后在我们的handleFinished() 中检查我们的状态。
private handleFinished(): void {
if (this.finished) {
// merge stream has finished, so nothing to check
return;
}
if (this.isStateFinished(this.mergingState)) {
this.stateCallbackAndSet(this.mergingState, null);
// set our mergingStream to null, to indicate we need a new one
// which will be fetched by getNextMergingIndex()
this.mergingStream = null;
this.mergeNextTick();
}
if (this.isStateFinished(this.writingState)) {
this.stateCallbackAndSet(this.writingState, null);
this.handleMainFinish(); // checks if there are still mergingStreams left, and sets finished flag
this.mergeNextTick();
}
}
isStateFinished() 检查我们的状态是否设置了终结标志以及队列大小是否等于 0
/**
* Method to check if a specific state has completed
* @param state the state to check
* @returns true if the state has completed
*/
private isStateFinished(state: MergingState): boolean {
if (!state || !state.finalizing || state.size > 0) {
return false;
}
return true;
}
一旦我们的结束回调位于合并Writable 流的最终回调中，就会设置最终标志。对于我们的主流，我们必须稍微不同地处理它，因为我们几乎无法控制我们的流何时结束，因为默认情况下可读调用我们可写的结束。我们想删除这种行为，以便我们可以决定何时结束我们的流。当设置其他最终侦听器时，这可能会导致一些问题，但对于大多数用例来说，这应该没问题。
private onPipe(readable: Readable): void {
// prevent our stream from being closed prematurely and unpipe it instead
readable.removeAllListeners('end'); // Note: will cause issues if another end listener is set
readable.once('end', () => {
this.finalizeState(this.writingState);
readable.unpipe();
});
}
finalizeState() 设置标志和回调以结束流。
/**
* Method to put a state in finalizing mode
*
* Finalizing mode: the last chunk has been received, when size is 0
* the stream should be removed.
*
* @param state the state which should be put in finalizing mode
*
*/
private finalizeState(state: MergingState, cb?: StreamCallback): void {
state.finalizing = true;
this.stateCallbackAndSet(state, cb);
this.mergeNextTick();
}
这就是您将多个流合并到一个接收器中的方式。
TL；DR：The complete code
此代码已使用我的 jest 测试套件在多个边缘案例上进行了全面测试，并且具有比我的代码中解释的更多的功能。例如附加流并合并到附加的流中。通过提供Merge.END 作为索引。
测试结果
你可以在这里看到我运行的测试，如果我忘记了，给我发消息，我可能会为它写另一个测试
MergeStream
✓ should throw an error when nextStream is not implemented (9ms)
✓ should throw an error when nextStream returns a stream with lower index (4ms)
✓ should reset index after new main stream (5ms)
✓ should write a single stream normally (50ms)
✓ should be able to merge a stream (2ms)
✓ should be able to append a stream on the end (1ms)
✓ should be able to merge large streams into a smaller stream (396ms)
✓ should be able to merge at the correct index (2ms)
用法
const mergingStream = new Merge({
*nextStream(): IterableIterator<MergingStream> {
for (let i = 0; i < 10; i++) {
const stream = new Readable();
stream.push(i.toString());
stream.push(null);
yield {index: i * 2, stream};
}
},
});
const template = new Readable();
template.push(', , , , , , , , , ');
template.push(null);
template.pipe(mergingStream).pipe(getSink());
我们的 sink 的结果将是
0, 1, 2, 3, 4, 5, 6, 7, 8, 9
最后的想法
这不是最省时的方法，因为我们一次只管理一个合并缓冲区。所以有很多等待。对于我的用例，这很好。我关心它不会占用我的记忆，这个解决方案对我有用。但肯定有一些优化的空间。完整的代码有一些额外的特性在这里没有完全解释，例如附加流和合并到附加的流中。不过他们已经用 cmets 解释过了。