Vertiefendes Verständnis der 4 Arten von Streams in Node.js

In diesem Artikel erfahren Sie, wie Sie die 4 Arten von Streams in Node.js verstehen und wie Sie das „Rückstau“-Problem des Pufferpuffers lösen können. ~

Setzen Sie einen ein Wie bringt man Dinge von A nach B?

Heben Sie es einfach an, bewegen Sie es zum Ziel und legen Sie es ab.

Was ist, wenn dieses Ding eine Tonne wiegt?

Dann verschieben Sie es Stück für Stück.

Tatsächlich bedeutet IO das Verschieben von Dingen, einschließlich Netzwerk-IO und Datei-IO. Wenn die Datenmenge gering ist, kann der gesamte Inhalt direkt übertragen werden. Wenn jedoch viel Inhalt vorhanden ist, wird er gleichzeitig in den Speicher geladen Die Zeit wird abstürzen und die Geschwindigkeit wird langsam sein. Dies ist die Idee des Flusses. [Empfohlenes Lernen: „nodejs Tutorial“]

Verschiedene Sprachen haben die Stream-API grundsätzlich implementiert, und Node.js hat auch die Stream-API implementiert, die am häufigsten verwendet wird.

Dieser Artikel beantwortet die folgenden Fragen:

• Was sind die 4 Streams von Node.js?
• Wie verbindet sich der Generator mit Readable Stream?
• Die Pause und der Fluss von Streams.
• Was ist das Gegendruckproblem und wie um es zu lösen

Die vier Arten von Streams in Node.js

Das intuitive Gefühl von Streams

Es gibt einen Abfluss und einen Zufluss. Der Abfluss ist ein lesbarer Stream (lesbar) und der Zufluss Eine Seite ist der beschreibbare Strom (beschreibbar).

Natürlich gibt es auch Streams, die sowohl ein- als auch ausfließen können. Dies wird als Duplex-Flow bezeichnet. Können wir den eingehenden Inhalt konvertieren, da er ein- und ausfließen kann? Diese Art von Fluss wird als Transformationsfluss (Transformation) bezeichnet

Duplex. Der Zufluss- und Abflussinhalt des Duplexflusses muss nicht zusammenhängen, während der Zufluss und der Abfluss des Transformationsflusses zusammenhängen Unterschied zwischen den beiden.

Stream-API

Die von Node.js bereitgestellten Streams sind die 4 oben vorgestellten Typen:

const stream = require('stream');

// 可读流
const Readable = stream.Readable;
// 可写流
const Writable = stream.Writable;
// 双工流
const Duplex = stream.Duplex;
// 转换流
const Transform = stream.Transform;

Sie alle verfügen über zu implementierende Methoden:

Readable muss die _read-Methode implementieren, um Inhalte zurückzugeben.

Writable-Anforderungen Implementieren Sie die Methode _write, um Inhalte zu akzeptieren.

Duplex muss die Methoden _read und _write implementieren, um Inhalte zu akzeptieren und zurückzugeben.
• Transform muss die Methode _transform implementieren, um den empfangenen Inhalt zu konvertieren und zurückzugeben. Schauen wir uns die einzelnen Methoden einmal an:
• Readable
• Readable muss die _read-Methode implementieren und bestimmte Daten per Push zurückgeben.

const Stream = require('stream');

const readableStream = Stream.Readable();

readableStream._read = function() {
    this.push('阿门阿前一棵葡萄树，');
    this.push('阿东阿东绿的刚发芽，');
    this.push('阿东背着那重重的的壳呀，');
    this.push('一步一步地往上爬。')
    this.push(null);
}

readableStream.on('data', (data)=> {
    console.log(data.toString())
});

readableStream.on('end', () => {
    console.log('done~');
});

Wenn eine Null gedrückt wird, bedeutet dies das Ende des Streams.

Der Ausführungseffekt ist wie folgt:

Das Erstellen von Readable kann auch durch Vererbung erfolgen:

const Stream = require('stream');

class ReadableDong extends Stream.Readable {

    constructor() {
        super();
    }

    _read() {
        this.push('阿门阿前一棵葡萄树，');
        this.push('阿东阿东绿的刚发芽，');
        this.push('阿东背着那重重的的壳呀，');
        this.push('一步一步地往上爬。')
        this.push(null);
    }

}

const readableStream = new ReadableDong();

readableStream.on('data', (data)=> {
    console.log(data.toString())
});

readableStream.on('end', () => {
    console.log('done~');
});

Der lesbare Stream generiert Inhalte, sodass er natürlich mit dem Generator kombiniert werden kann:

const Stream = require('stream');

class ReadableDong extends Stream.Readable {

    constructor(iterator) {
        super();
        this.iterator = iterator;
    }

    _read() {
        const next = this.iterator.next();
        if(next.done) {
            return this.push(null);
        } else {
            this.push(next.value)
        }
    }

}

function *songGenerator() {
    yield '阿门阿前一棵葡萄树，';
    yield '阿东阿东绿的刚发芽，';
    yield '阿东背着那重重的的壳呀，';
    yield '一步一步地往上爬。';
}

const songIterator = songGenerator();

const readableStream = new ReadableDong(songIterator);

readableStream.on('data', (data)=> {
    console.log(data.toString())
});

readableStream.on('end', () => {
    console.log('done~');
});

Dies ist der lesbare Stream , durch Implementieren Sie die _read-Methode, um Inhalte zurückzugeben.

Writable

Writable Um die _write-Methode zu implementieren, erhalten Sie den geschriebenen Inhalt.

const Stream = require('stream');

const writableStream = Stream.Writable();

writableStream._write = function (data, enc, next) {
   console.log(data.toString());
   // 每秒写一次
   setTimeout(() => {
       next();
   }, 1000);
}

writableStream.on('finish', () => console.log('done~'));

writableStream.write('阿门阿前一棵葡萄树，');
writableStream.write('阿东阿东绿的刚发芽，');
writableStream.write('阿东背着那重重的的壳呀，');
writableStream.write('一步一步地往上爬。');
writableStream.end();

Empfängt den geschriebenen Inhalt, druckt ihn aus und ruft als nächstes auf, um den nächsten geschriebenen Inhalt zu verarbeiten. Der nächste Aufruf erfolgt hier asynchron und die Häufigkeit kann gesteuert werden.

Nach einer Weile kann der geschriebene Inhalt tatsächlich normal verarbeitet werden:

Dies ist ein beschreibbarer Stream, und der geschriebene Inhalt wird durch Implementierung der _write-Methode verarbeitet.

Duplex

Duplex ist lesbar und beschreibbar, implementieren Sie einfach _read und _write gleichzeitig

const Stream = require('stream');

var duplexStream = Stream.Duplex();

duplexStream._read = function () {
    this.push('阿门阿前一棵葡萄树，');
    this.push('阿东阿东绿的刚发芽，');
    this.push('阿东背着那重重的的壳呀，');
    this.push('一步一步地往上爬。')
    this.push(null);
}

duplexStream._write = function (data, enc, next) {
    console.log(data.toString());
    next();
}

duplexStream.on('data', data => console.log(data.toString()));
duplexStream.on('end', data => console.log('read done~'));

duplexStream.write('阿门阿前一棵葡萄树，');
duplexStream.write('阿东阿东绿的刚发芽，');
duplexStream.write('阿东背着那重重的的壳呀，');
duplexStream.write('一步一步地往上爬。');
duplexStream.end();

duplexStream.on('finish', data => console.log('write done~'));

Integrieren Sie die Funktionen von lesbarem Stream und beschreibbarem Stream, das ist Duplex-Stream Duplex.

Transformieren

Obwohl der Duplex-Stream gelesen und geschrieben werden kann, besteht manchmal keine Verbindung zwischen den beiden. In diesem Fall muss der Stream transformiert werden transformiert werden.

Der Transformationsstream muss die _transform-API implementieren. Wir implementieren den Transformationsstream, der den Inhalt umkehrt:

const Stream = require('stream');

class TransformReverse extends Stream.Transform {

  constructor() {
    super()
  }

  _transform(buf, enc, next) {
    const res = buf.toString().split('').reverse().join('');
    this.push(res)
    next()
  }
}

var transformStream = new TransformReverse();

transformStream.on('data', data => console.log(data.toString()))
transformStream.on('end', data => console.log('read done~'));

transformStream.write('阿门阿前一棵葡萄树');
transformStream.write('阿东阿东绿的刚发芽');
transformStream.write('阿东背着那重重的的壳呀');
transformStream.write('一步一步地往上爬');
transformStream.end()

transformStream.on('finish', data => console.log('write done~'));

Führen Sie ihn eine Weile aus. Der Effekt ist wie folgt:

流的暂停和流动

我们从 Readable 流中获取内容，然后流入 Writable 流，两边分别做 _read 和 _write 的实现，就实现了流动。

背压

但是 read 和 write 都是异步的，如果两者速率不一致呢？

如果 Readable 读入数据的速率大于 Writable 写入速度的速率，这样就会积累一些数据在缓冲区，如果缓冲的数据过多，就会爆掉，会丢失数据。

而如果  Readable 读入数据的速率小于 Writable 写入速度的速率呢？那没关系，最多就是中间有段空闲时期。

这种读入速率大于写入速率的现象叫做“背压”，或者“负压”。也很好理解，写入段压力比较大，写不进去了，会爆缓冲区，导致数据丢失。

这个缓冲区大小可以通过 readableHighWaterMark 和 writableHightWaterMark 来查看，是 16k。

解决背压

怎么解决这种读写速率不一致的问题呢？

当没写完的时候，暂停读就行了。这样就不会读入的数据越来越多，驻留在缓冲区。

readable stream 有个 readableFlowing 的属性，代表是否自动读入数据，默认为 true，也就是自动读入数据，然后监听 data 事件就可以拿到了。

当 readableFlowing 设置为 false 就不会自动读了，需要手动通过 read 来读入。

readableStream.readableFlowing = false;

let data;
while((data = readableStream.read()) != null) {
    console.log(data.toString());
}

但自己手动 read 比较麻烦，我们依然可以用自动流入的方式，调用 pause 和 resume 来暂停和恢复就行了。

当调用 writable stream 的 write 方法的时候会返回一个 boolean 值代表是写入了目标还是放在了缓冲区：

• true: 数据已经写入目标
• false：目标不可写入，暂时放在缓冲区

我们可以判断返回 false 的时候就 pause，然后等缓冲区清空了就 resume：

const rs = fs.createReadStream(src);
const ws = fs.createWriteStream(dst);

rs.on('data', function (chunk) {
    if (ws.write(chunk) === false) {
        rs.pause();
    }
});

rs.on('end', function () {
    ws.end();
});

ws.on('drain', function () {
    rs.resume();
});

这样就能达到根据写入速率暂停和恢复读入速率的功能，解决了背压问题。

pipe 有背压问题么？

平时我们经常会用 pipe 来直接把 Readable 流对接到 Writable 流，但是好像也没遇到过背压问题，其实是 pipe 内部已经做了读入速率的动态调节了。

const rs = fs.createReadStream(src);
const ws = fs.createWriteStream(dst);

rs.pipe(ws);

总结

流是传输数据时常见的思想，就是一部分一部分的传输内容，是文件读写、网络通信的基础概念。

Node.js 也提供了 stream 的 api，包括 Readable 可读流、Writable 可写流、Duplex 双工流、Transform 转换流。它们分别实现 _read、_write、_read + _write、_transform 方法，来做数据的返回和处理。

创建 Readable 对象既可以直接调用 Readable api 创建，然后重写 _read 方法，也可以继承 Readable 实现一个子类，之后实例化。其他流同理。（Readable 可以很容易的和 generator 结合）

当读入的速率大于写入速率的时候就会出现“背压”现象，会爆缓冲区导致数据丢失，解决的方式是根据 write 的速率来动态 pause 和 resume 可读流的速率。pipe 就没有这个问题，因为内部做了处理。

流是掌握 IO 绕不过去的一个概念，而背压问题也是流很常见的问题，遇到了数据丢失可以考虑是否发生了背压。希望这篇文章能够帮大家理清思路，真正掌握 stream！

更多编程相关知识，请访问：编程入门！！

Das obige ist der detaillierte Inhalt vonVertiefendes Verständnis der 4 Arten von Streams in Node.js. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!