Node.js のストリームを理解する — 効率的なデータ処理

ストリームは Node.js の強力な機能で、すべてを一度にメモリにロードするのではなく、大量のデータを少しずつ処理することで効率的に処理できます。これらは、大きなファイル、リアルタイム データ、さらにはネットワーク接続を処理する場合に特に役立ちます。この記事では、Node.js ストリームについて詳しく説明し、ストリームの種類、ストリームの使用方法とコード例、および理解を確実にするための実際の使用例を取り上げます。

ストリームとは何ですか?

ストリームは、時間の経過とともに処理される一連のデータです。 Node.js では、ストリームは EventEmitter のインスタンスであり、イベントを発行してイベントに応答できることを意味します。ストリームを使用すると、すべてのデータを一度にロードするのではなく、データをチャンク (小さな断片) で読み書きできるため、メモリ効率が高く、高速になります。

ストリームを使用する理由

• 効率的なメモリ使用量: ストリームは、データ セット全体をメモリにロードすることなく、データをチャンクで処理します。
• 処理の高速化: すべてが読み込まれるのを待つのではなく、利用可能になるとすぐにデータの処理を開始します。
• ノンブロッキング I/O: ストリームは非同期で動作するため、他の操作をブロックせず、リアルタイム アプリケーションに最適です。

ストリームの種類

Node.js は 4 種類のストリームを提供します。

1. 読み取り可能なストリーム: データを順番に読み取るために使用されます。
2. 書き込み可能なストリーム: データを順次書き込むために使用されます。
3. 二重ストリーム: 読み取りと書き込みの両方が可能です。
4. 変換ストリーム: 出力が入力の変換である二重ストリーム。

例を挙げて、各タイプのストリームを見てみましょう。

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

読み取り可能なストリームを使用すると、ファイルやネットワーク リクエストなどのソースからデータをチャンクごとに消費できます。

例: 読み取り可能なストリームを使用したファイルの読み取り

const fs = require('fs');

// Create a readable stream
const readableStream = fs.createReadStream('example.txt', 'utf8');

// Listen for 'data' events to read chunks of data
readableStream.on('data', (chunk) => {
  console.log('New chunk received:');
  console.log(chunk);
});

// Handle 'end' event when the file has been completely read
readableStream.on('end', () => {
  console.log('File reading completed.');
});

// Handle any errors
readableStream.on('error', (err) => {
  console.error('Error reading file:', err.message);
});

説明:

• fs.createReadStream() は、example.txt の内容を読み取るストリームを作成します。
• ストリームは、読み取ったチャンクごとに「data」イベントを発行し、読み取りが終了すると「end」イベントを発行します。

書き込み可能なストリーム

書き込み可能なストリームは、データをファイルに保存するなど、データをチャンクごとに書き込むために使用されます。

例: 書き込み可能なストリームを使用してファイルにデータを書き込む

const fs = require('fs');

// Create a writable stream
const writableStream = fs.createWriteStream('output.txt');

// Write chunks of data to the file
writableStream.write('First chunk of data.\n');
writableStream.write('Second chunk of data.\n');

// End the stream
writableStream.end('Final chunk of data.');

// Handle 'finish' event when writing is complete
writableStream.on('finish', () => {
  console.log('Data writing completed.');
});

// Handle any errors
writableStream.on('error', (err) => {
  console.error('Error writing to file:', err.message);
});

説明:

• fs.createWriteStream() は、output.txt に書き込む書き込み可能なストリームを作成します。
• write() メソッドは、データのチャンクをストリームに送信するために使用されます。すべてのデータが書き込まれると、end() メソッドが呼び出され、ストリームの終了を通知します。

二重ストリーム

二重ストリームはデータの読み取りと書き込みの両方が可能で、データの送受信が必要なネットワーク プロトコルなどの操作に使用されます。

例: カスタム二重ストリーム

const { Duplex } = require('stream');

// Create a custom duplex stream
const myDuplexStream = new Duplex({
  read(size) {
    this.push('Reading data...');
    this.push(null);  // No more data to read
  },
  write(chunk, encoding, callback) {
    console.log(`Writing: ${chunk.toString()}`);
    callback();
  }
});

// Read from the stream
myDuplexStream.on('data', (chunk) => {
  console.log(chunk.toString());
});

// Write to the stream
myDuplexStream.write('This is a test.');
myDuplexStream.end();

説明:

• 二重ストリームは、読み取り操作と書き込み操作の両方を実行できます。この例では、二重ストリームのカスタム読み取りおよび書き込みメソッドを定義します。

ストリームの変換

変換ストリームを使用すると、データが通過するときにデータを変更または変換できます。これらは特殊なタイプの二重ストリームです。

例: 大文字テキストへの単純な変換ストリーム

const { Transform } = require('stream');

// Create a custom transform stream
const toUpperCaseTransform = new Transform({
  transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase());
    callback();
  }
});

// Pipe data through the transform stream
process.stdin.pipe(toUpperCaseTransform).pipe(process.stdout);

説明:

• 変換ストリームは入力を受け取り、それを処理し (この場合、テキストを大文字に変換し)、変更されたデータを出力します。
• この例では、データは変換ストリームを介して標準入力 (process.stdin) からパイプされ、結果はコンソール (process.stdout) に出力されます。

配管ストリーム

ストリームを操作する最も一般的な方法の 1 つは、ストリームを一緒に「パイプ」することです。これは、あるストリームから別のストリームにデータを渡すことを意味します。これは、ファイルからの読み取りや別のファイルへの書き込みなど、データを段階的に処理する必要がある場合に便利です。

例: 読み取り可能なストリームを書き込み可能なストリームにパイプする

const fs = require('fs');

// Create a readable stream
const readableStream = fs.createReadStream('input.txt');

// Create a writable stream
const writableStream = fs.createWriteStream('output.txt');

// Pipe the readable stream into the writable stream
readableStream.pipe(writableStream);

// Handle 'finish' event when piping is done
writableStream.on('finish', () => {
  console.log('File copied successfully.');
});

説明:

• pipe() メソッドは、読み取り可能なストリーム (input.txt) から書き込み可能なストリーム (output.txt) にデータを直接渡します。

実際の使用例: 大きなファイルのアップロードをストリーミングする

実際のアプリケーションでは、大きなファイルをサーバーにアップロードする必要がある場合があります。ファイル全体をメモリにロードする代わりに、ストリームを使用してファイルのアップロードを効率的に処理できます。

例: Node.js と multer でストリームを使用してファイルをアップロードする

const express = require('express');
const multer = require('multer');
const fs = require('fs');

const app = express();
const upload = multer({ dest: 'uploads/' });

app.post('/upload', upload.single('file'), (req, res) => {
  const readableStream = fs.createReadStream(req.file.path);
  const writableStream = fs.createWriteStream(`./uploads/${req.file.originalname}`);

  // Pipe the uploaded file to the writable stream
  readableStream.pipe(writableStream);

  writableStream.on('finish', () => {
    res.send('File uploaded and saved.');
  });

  writableStream.on('error', (err) => {
    res.status(500).send('Error saving file.');
  });
});

app.listen(3000, () => {
  console.log('Server is running on port 3000');
});

Explanation:

• We use multer to handle file uploads. When the file is uploaded, it is piped from a temporary location to the desired directory on the server.
• This method is efficient as it streams the file data instead of holding it all in memory at once.

Best Practices for Working with Streams

1. Error Handling: Always handle errors in streams to avoid unhandled exceptions, especially when dealing with file systems or network operations.

Example:

   readableStream.on('error', (err) => {
     console.error('Stream error:', err.message);
   });

1. Flow Control: Be mindful of flow control when reading and writing data, as writable streams can become overwhelmed if data is being written faster than it can be consumed.

Example:

   writableStream.write(chunk, (err) => {
     if (err) console.error('Error writing chunk:', err.message);
   });

1. Use Pipe for Simplicity: When transferring data between streams, always prefer using pipe() instead of manually managing the flow of data.

Conclusion

Streams in Node.js offer a powerful and efficient way to handle data, especially in cases where data comes in large quantities or needs to be processed incrementally. From reading and writing files to handling network requests and processing data in real time, streams allow you to build scalable and performant applications. In this article, we explored the different types of streams, how to use them, and real-world use cases to deepen your understanding of stream-based processing in Node.js.

以上がNode.js のストリームを理解する — 効率的なデータ処理の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。