Node.js でのストリーム読み取り可能なストリームの使用-jsチュートリアル-php.cn

読み取り可能なストリームとは、プログラムで使用するためのデータを生成するストリームです。一般的なデータ生成方法には、ディスクファイルの読み取り、ネットワークリクエストコンテンツの読み取りなどが含まれます。ストリームとは何かについての前の例を見てください。

const rs = fs.createReadStream(filePath);

ログイン後にコピー

rs は読み取り可能なストリームであり、そのデータ生成方法は次のとおりです。 read ディスクファイル、コンソール process.stdin も読み取り可能なストリームです:

process.stdin.pipe(process.stdout);

ログイン後にコピー

簡単な文でコンソール入力を出力できます。 process.stdin がデータを生成する方法は、コンソールでのユーザーの入力を読み取ることです。の入力です。

読み取り可能なストリームの定義を振り返ってください。

読み取り可能なストリームは、プログラムで使用するデータを生成するストリームです。

カスタム読み取り可能なストリーム

提供されるシステムに加えて

fs.CreateReadStream

使用される gulp またはヴァイナル-fs によって提供される src メソッドも読み取り可能なストリームを使用します

gulp.src(['*.js', 'dist/**/*.scss'])

ログイン後にコピー

特定の方法でデータを生成し、それをプログラムに渡して消費したい場合、どのように始めればよいでしょうか?

これは 2 つの簡単な手順で実行できます。

Readable
クラスを継承します。
_read
メソッド、this.push
を呼び出して、生成されたデータをキューに入れて読み取ります

例: 100 ミリ秒ごとに乱数を生成するストリームを実装します (役に立ちません)

const Readable = require('stream').Readable;
class RandomNumberStream extends Readable {
    constructor(max) {
        super()
    }
    _read() {
        const ctx = this;
        setTimeout(() => {
            const randomNumber = parseInt(Math.random() * 10000);
            // 只能 push 字符串或 Buffer，为了方便显示打一个回车
            ctx.push(`${randomNumber}\n`);
        }, 100);
    }
}
module.exports = RandomNumberStream;

ログイン後にコピー

コードのクラス継承部分は非常に単純です。主に、 _read メソッドには、いくつかの注意すべき点があります。

Readable クラスにはデフォルトで _read メソッドが実装されていますが、何も行われません。 #_read メソッドにはパラメータ size があり、どれだけのデータを読み取って read メソッドに返すかを指定するために使用されますが、これは単なる参照データです。多くの実装ではこのパラメータが無視され、ここでも無視されます。詳細は後述

これを介してバッファにプッシュするデータをプッシュする、バッファの概念は後述しますが、とりあえず水道管に押し込んで消費できることが分かります
プッシュのコンテンツは文字列またはバッファのみであり、数値は使用できません。
プッシュメソッドには次のパラメータがあります。最初のパラメータが文字列の場合、2 番目のパラメータエンコーディングはエンコーディングを指定するために使用されます。

const RandomNumberStream = require('./RandomNumberStream');
const rns = new RandomNumberStream();
rns.pipe(process.stdout);

ログイン後にコピー

停止方法

100 ミリ秒ごとに数値をバッファにプッシュすると、たとえば、ローカルファイルを読み取る場合、常に終了します。データが読み取られたことを示しますか?

null をバッファにプッシュするだけです。消費者が必要な乱数の数を定義できるようにコードを変更します:

const Readable = require('stream').Readable;
class RandomNumberStream extends Readable {
    constructor(max) {
        super()
        this.max = max;
    }
    _read() {
        const ctx = this;
        setTimeout(() => {
            if (ctx.max) {
                const randomNumber = parseInt(Math.random() * 10000);
                // 只能 push 字符串或 Buffer，为了方便显示打一个回车
                ctx.push(`${randomNumber}\n`);
                ctx.max -= 1;
            } else {
                ctx.push(null);
            }
        }, 100);
    }
}
module.exports = RandomNumberStream;

ログイン後にコピー

コードは最大識別子を使用して、消費者がニーズを指定できるようにします。インスタンス化時に文字数を指定できます

const RandomNumberStream = require('./');
const rns = new RandomNumberStream(5);
rns.pipe(process.stdout);

ログイン後にコピー

このようにすると、コンソールに 5 文字しか出力されないことがわかります

なぜ setInterval ではなく setTimeout なのでしょうか

注意してください学生は、100 ミリ秒ごとの乱数の生成では setInterval を呼び出すのではなく、setTimeout を使用することに気づいたかもしれませんが、結果が正しいのに、遅延するだけで繰り返されないのはなぜでしょうか。

#これには、ストリームが機能する 2 つの方法を理解する必要があります

#フローモード: データは基礎となるシステムによって読み取られ、できるだけ早くアプリケーションに提供されます

一時停止モード: 複数のデータブロックを読み取るには、read() メソッドを明示的に呼び出す必要があります

フローモードと一時停止モードの切り替え
次のメソッドを使用して、ストリームをデフォルトの一時停止モードからフローに切り替えることができます。 mode:

データイベントリスナーを追加してデータ監視を開始する

resume() メソッドを呼び出してデータフローを開始する

pipe() メソッドを呼び出して、データを別の書き込み可能なストリームに転送します

フローモードから一時停止モードに切り替える方法は 2 つあります。
ストリームに Pipe() がない場合、 stop() メソッドはストリームを一時停止できます

pipe() すべてのデータイベントリスナーを削除し、unpipe() メソッドを呼び出します

#data event

使用後Pipe() メソッドを使用すると、データは読み取り可能なストリームからストリーミングされます書き込み可能なストリームに入りましたが、ユーザーにとってはブラックボックスに見えますデータはどのように流れるのでしょうか?フローモードと一時停止モードを切り替えるときに重要な用語が 2 つあります。
#フローモードに対応するデータイベント

一時停止モードに対応する read() メソッド

これら 2 つのメカニズムは、プログラムがデータフローを駆動できる理由です。最初にフローモードのデータイベントを見てみましょう。読み取り可能なストリームのデータイベントが監視されると、ストリームはフローモードに入ります。上記のストリームを呼び出すコードを書き換えることができます。

const RandomNumberStream = require('./RandomNumberStream');
const rns = new RandomNumberStream(5);
rns.on('data', chunk => {
  console.log(chunk);
});

ログイン後にコピー

这样可以看到控制台打印出了类似下面的结果

<Buffer 39 35 37 0a>
<Buffer 31 30 35 37 0a>
<Buffer 38 35 31 30 0a>
<Buffer 33 30 35 35 0a>
<Buffer 34 36 34 32 0a>

ログイン後にコピー

当可读流生产出可供消费的数据后就会触发 data 事件，data 事件监听器绑定后，数据会被尽可能地传递。data 事件的监听器可以在第一个参数收到可读流传递过来的 Buffer 数据，这也就是控制台打印的 chunk，如果想显示为数字，可以调用 Buffer 的 toString() 方法

当数据处理完成后还会触发一个

end

事件，因为流的处理不是同步调用，所以如果希望完事后做一些事情就需要监听这个事件，在代码最后追加一句：

rns.on('end', () => {
  console.log('done');
});复制代码

ログイン後にコピー

这样可以在数据接收完了显示 done ，当然数据处理过程中出现了错误会触发 error 事件，可以监听做异常处理：

rns.on('error', (err) => {
  console.log(err);
});复制代码

ログイン後にコピー

read(size)

流在暂停模式下需要程序显式调用 read() 方法才能得到数据，read() 方法会从内部缓冲区中拉取并返回若干数据，当没有更多可用数据时，会返回null

使用 read() 方法读取数据时，如果传入了 size 参数，那么它会返回指定字节的数据；当指定的size字节不可用时，则返回null。如果没有指定size参数，那么会返回内部缓冲区中的所有数据

现在有一个矛盾，在流动模式下流生产出了数据，然后触发 data 事件通知给程序，这样很方便。在暂停模式下需要程序去读取，那么就有一种可能是读取的时候还没生产好，如果使用轮询的方式未免效率有些低

NodeJS 提供了一个

readable的事件，事件在可读流准备好数据的时候触发，也就是先监听这个事件，收到通知有数据了再去读取就好了：

const rns = new RandomNumberStream(5);
rns.on('readable', () => {
  let chunk;
  while((chunk = rns.read()) !== null){
    console.log(chunk);
  }
});

ログイン後にコピー

这样可以读取到数据，值得注意的一点是并不是每次调用 read() 方法都可以返回数据，前面提到了如果可用的数据没有达到 size 那么返回 null，所以在程序中加了个判断

数据会不会漏掉

const stream = fs.createReadStream('/dev/input/event0');
stream.on('readable', callback);复制代码

ログイン後にコピー

在流动模式会不会有这样的问题：可读流在创建好的时候就生产数据了，如果在绑定 readable 事件之前就生产了某些数据，触发了 readable 事件，在极端情况下会造成数据丢失吗？

事实并不会，按照 NodeJS event loop 程序创建流和调用事件监听在一个事件队列里面，生产数据和事件监听都是异步操作，而 on 监听事件使用了 process.nextTick 会保证在数据生产之前被绑定好，相关知识可以看定时器章节中对 event loop 的解读

到这里可能对 data事件、readable事件触发时机， read() 方法每次读多少数据，什么时候返回 null 还有一定的疑问，在后续可写流章节会在 back pressure 部分结合源码介绍相关机制

推荐教程：《JS教程》

以上がNode.js でのストリーム読み取り可能なストリームの使用の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。