Node.js_node.js のストリームの概要

フローとは何ですか?

ストリームと言えば、*nix の概念が含まれます。 Pipe - *nix では、ストリームは (パイプ文字) を介してブリッジできるデータとしてシェルに実装されます。プロセス (stdout) は、次のプロセスの入力 (stdin) として直接使用できます。

Node では、ストリーム (Stream) の概念が似ており、データ ストリームのブリッジ機能を表します。

パイプ

ストリーミングの本質は .pipe() メソッドにあります。ブリッジ機能は、データ ストリームの両端 (アップストリーム/ダウンストリーム、または読み取り/書き込みストリーム) が .pipe() メソッドでブリッジされることです。

擬似コードの表現形式は次のとおりです:

コードをコピー コードは次のとおりです:

//上流.pipe (下流)
Readable.pipe(Writable);

ストリームの分類

これは、Node v0.4 より前のいわゆる「クラシック」フローについて説明することを目的としたものではありません。次に、ストリームはいくつかのカテゴリ (すべて抽象インターフェイス:

) に分類されます。

1.stream.Readable 読み取り可能なストリーム (_read メソッドを実装する必要があります。焦点はデータ ストリームの読み取りの詳細にあります
) 2.stream.Writable 書き込み可能なストリーム (_write メソッドを実装する必要があります。焦点はデータ ストリーム
の書き込みの詳細にあります) 3.stream.Duplex 読み取り/書き込みストリーム (上記の 2 つのインターフェイスを実装する必要があります。上記の 2 つのインターフェイスの詳細に注目してください
4.stream.Transform Duplex から継承 (_transform メソッドを実装する必要があります。焦点はデータ ブロックの処理です

)

簡単に言うと:

1) .pipe() の所有者は、一連の「readable」/「data」/「end」/「close」/「error」イベントを持つ読み取り可能なストリーム (ただし、これに限定されない) 機能を持っている必要があります。サブスクリプションは、
を呼び出すための .read()/.pause()/.resume() などの一連のメソッドも提供します。 2) .pipe() のパラメータには書き込み可能なストリーム機能が必要です (ただし、これに限定されません)。アクセス用の 'drain'/'pipe'/'unpipe'/'error'/'finish' イベントがあり、 も提供されます。

の呼び出しには、write()/.end() やその他のメソッドが利用可能です

一体何だ

少しでも不安を感じていませんか？心配しないでください。人間の言語を話す低レベルのプログラマーとして、私は Stream を分解して、それについてお話します。

Node.js ソース コード 内の Stream クラスは次のように定義されています:

コードをコピー コードは次のとおりです:

var EE = require('events').EventEmitter;
var util = require('util');
util.inherits(ストリーム、EE);

関数 Stream() {
EE.call(this);
}

ご覧のとおり、Stream は本質的に EventEmitter であり、イベント駆動型の関数 (.emit/.on...) があることを意味します。ご存知のとおり、「Node.js は V8 をベースとしたイベント駆動型プラットフォーム」であり、イベント駆動型のストリーミング プログラミングを実装し、Node と同じ非同期コールバック特性を備えています。

たとえば、読み取り可能ストリームには読み取り可能イベントがあり、一時停止された読み取り専用ストリームでは、データ ブロックを読み取る準備ができている限り、サブスクライバーに送信されます (読み取り可能ストリームとは何ですか)。 ? Express) req、ftp または複数フォームアップロードコンポーネントの req.part、システムの標準入力プロセス.stdin など)。読み取り可能なイベントを使用すると、シェル コマンドの出力を処理するパーサーなどのツールを作成できます:

コードをコピー コードは次のとおりです:

process.stdin.on('読み取り可能', function(){
var buf = process.stdin.read();
if(buf){
var data = buf.toString();
// データを解析しています ... }
});

次のように呼び出します:

コードをコピーします コードは次のとおりです:

head -10 some.txt ノード parser.js

Readable ストリームの場合、古典的なソケットの例 のように、そのデータと終了イベントをサブスクライブしてデータのチャンクを取得し、ストリームが使い果たされたときに通知を受け取ることもできます。

コードをコピー コードは次のとおりです:

req.on('connect', function(res, ソケット, ヘッド) {
ソケット.on('データ', 関数(チャンク) {
console.log(chunk.toString());
});
ソケット.on('end', function() {
proxy.close();
});
});

読み取り可能なストリームステータスの切り替え
Readable ストリームには、フロー モード (トレント) と一時停止モード (一時停止) の 2 つの状態があることに注意してください。前者はまったく停止できず、パイプされた相手にフィードを送り続けますが、後者はダウンストリームが明示的に Stream.read() リクエストを呼び出してデータ ブロックを読み取るまで一時停止します。 Readable ストリームは、初期化時に一時停止モードになります。

これら 2 つの状態は相互に切り替えることができます。

次のいずれかの動作が発生した場合、一時停止が流れに変わります:

1. データ イベント サブスクリプションを Readable ストリームに追加します

2. Readable で .resume() を呼び出して、フロー
を明示的に有効にします 3. Readable ストリームの .pipe(writable) を呼び出して、Writable ストリームにブリッジします

次の動作のいずれかが発生した場合、フローは一時停止に戻ります:

1.Readable ストリームはまだどのストリームにもパイプされていません。調整可能な .pause() を使用して一時停止できます

2. Readable ストリームはストリームにパイプされています。すべてのデータ イベント サブスクリプションを削除し、.unpipe() メソッドを呼び出して、ダウンストリーム ストリームとの関係を 1 つずつ解放する必要があります。

素晴らしい使い方

ストリームの非同期特性と組み合わせると、次のようなアプリケーションを作成できます。ユーザー A の出力をユーザー B のページの出力に直接ブリッジします。

コードをコピー コードは次のとおりです:

router.post('/post', function(req, res) {
var destination = req.headers['destination'] //送信先
; キャッシュ[宛先] = req;
//はい、返されないので、ajax リクエストを行うのが最善です
});

ユーザー B がリクエストした場合:

コードをコピーします コードは次のとおりです:

router.get('/inbox', function(req, res){
var user = req.headers['user'];
cache.find(user, function(err,previousReq){ //以前に保存した req を検索
var form = new multiparty.Form();
form.parse(previousReq) // 私用のファイルがあります
form.on('part', function (part) {
part.pipe(res) //ストリーミングメソッドは良いです:)

part.on('error', function (err) {
console.log(err);
messageging.setRequestDone(uniqueID);
return res.end(err);
});
});
});
});

参考

ストリームを使用したノード プログラムの作成方法: ストリーム ハンドブック