Node.js の 4 つの主要なフローを深く理解し、「バック プレッシャー」問題を解決する

Node.js4 つのストリームとは何ですか?バックプレッシャーの問題とは何ですか?次の記事では、Node.js の 4 つの主要なストリームを理解し、バックプレッシャーの問題と解決策を紹介します。

何かを A から B に移動するにはどうすればよいでしょうか?

持ち上げて目的地まで移動し、置くだけ。

もしこれが1トンの重さだったらどうなるでしょうか?

次に、部分ごとに移動します。

実際には、IO とはネットワーク IO やファイル IO などを移動させることを意味し、データ量が少なければすべてのコンテンツを直接転送できますが、コンテンツが多い場合は、コンテンツをロードする必要があります。メモリは一度にクラッシュするし、速度も遅いので部分的に処理できる、これがフローの考え方です。

ストリーム API は基本的にさまざまな言語で実装されており、Node.js でも実装されています。ストリーム API は比較的よく使われています。以下のストリームを見てみましょう。

この記事では、次の質問に答えます:

• Node.js の 4 つのストリームとは何ですか

• Node.js の 4 つのストリームとはどのようなものですか?ジェネレーターは Readable Stream と対話し、

• stream

• バック プレッシャーの問題とその解決方法

  の一時停止とフローを組み合わせます。

[推奨学習:「nodejs チュートリアル 」]

##Node.js の 4 種類のストリーム

直感的な流れ感

ある場所から別の場所への流れ、当然流出側と流入側があり、流出側はリーダブルストリーム(可読) 、流入側は書き込み可能です。

もちろん、流入と流出の両方が可能なストリームもあります。これは二重ストリームと呼ばれます

# #流入と流出が可能なので、受信コンテンツをダウンコンバートしてから流出することはできますか? この種のフローは変換フロー (transform) と呼ばれます

duplexフロー 流入コンテンツと流出コンテンツは関連している必要はありませんが、変換フローの流入と流出は関連している、これが 2 つの違いです。

Stream api

Node.js によって提供されるストリームは、上記で紹介した 4 種類です。

const stream = require('stream');

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

これらはすべてです。実装するメソッドがあります:

• Readable はコンテンツを返すために _read メソッドを実装する必要があります

• Writable はコンテンツを返すために _write メソッドを実装する必要がありますコンテンツを受け入れる

• Duplex はコンテンツを受け入れて返すために _read メソッドと _write メソッドを実装する必要があります

• Transform は変換するために _transform メソッドを実装する必要があります受信したコンテンツを返します。

これらを個別に見てみましょう:

Readable

Readable _read を実装するにはメソッドの場合は、 Push を通じて特定のデータを返します。

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~');
});

null をプッシュすると、ストリームの終わりを意味します。

実行結果は次のとおりです:

Readable の作成は継承を通じて行うこともできます:

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~');
});

Readable ストリームはコンテンツを生成します。次に、これをジェネレーターと組み合わせるのが自然です。

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~');
});

これは読み取り可能なストリームであり、コンテンツは _read メソッドを実装することによって返されます。

Writable

Writable は、書き込まれたコンテンツを受け取る _write メソッドを実装します。

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();

書き込まれた内容を受信し、出力し、次の書き込み内容を処理するために next を呼び出します。ここでの次の呼び出しは非同期であり、頻度は制御できます。

しばらく実行すると、書き込まれたコンテンツは実際に正常に処理できます。

これは書き込み可能なストリームであり、書き込みは_write メソッドの内容。

Duplex

Duplex は読み取りおよび書き込み可能です。_read と _write を同時に実装するだけです。

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~'));

Readable ストリームと Writable ストリーム関数を統合します。はデュプレックスです。

Transform

Duplex ストリームは読み取りと書き込みが可能ですが、この 2 つのストリームの間には関連性がなく、場合によっては関連性があります。必要な受信コンテンツを変換した後、流出するのですが、このときストリーム Transform を変換する必要があります。

変換ストリームは _transform API を実装する必要があります。コンテンツを反転する変換ストリームを実装します:

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~'));

実行後の効果は次のとおりです:

流的暂停和流动

我们从 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！

更多编程相关知识，请访问：编程视频！！

以上がNode.js の 4 つの主要なフローを深く理解し、「バック プレッシャー」問題を解決するの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。