【Node.jsストリーム処理の極意】Async/Awaitでイベントベース、Promise、Transform、Duplexを制覇！
JavaScript、Node.js、Async/Await を用いたストリーム処理の分かりやすい解説
このチュートリアルでは、JavaScript、Node.js、Async/Await を用いたストリーム処理について、分かりやすく解説します。
ストリームとは
ストリームは、データが連続的に流れるパイプのようなものです。データは一度にすべて渡されるのではなく、小さな塊（チャンク）に分割されて渡されます。これは、大きなファイルやネットワークデータなど、処理に時間がかかるデータを読み書きする場合に特に有効です。
Async/Await とは
Async/Await は、非同期処理を扱うための構文です。Promise を用いた従来の非同期処理と比較して、コードをより読みやすく、簡潔に記述することができます。
ES8 の async/await とストリーム
Node.js 10.0.0 以降では、ES8 の async/await を用いてストリームを処理することができます。これは、従来のイベントベースのコードよりも、より簡潔で読みやすいコードを書くことを可能にします。
ストリーム処理の例
以下の例では、fs.createReadStream 関数を使用してファイルを読み込み、その内容をコンソールに出力します。
async function readStream(fileName) {
const stream = fs.createReadStream(fileName);
for await (const chunk of stream) {
console.log(chunk.toString());
}
await stream.close();
}
readStream('example.txt');
このコードは、以下のようになります。
fs.createReadStream関数を使用して、ファイルを読み込むためのストリームを作成します。for await...ofループを使用して、ストリームからチャンクを順次読み取ります。- 各チャンクをコンソールに出力します。
await stream.close()を使用して、ストリームを閉じます。
パイプライン処理
ストリームは、パイプラインと呼ばれる方法で連結することができます。パイプライン処理を使用すると、複数のストリームを連結し、それぞれからデータを受け取って処理することができます。
以下の例では、fs.createReadStream 関数と fs.createWriteStream 関数を使用して、ファイルを別のファイルにコピーします。
async function copyFile(sourceFileName, destFileName) {
const readStream = fs.createReadStream(sourceFileName);
const writeStream = fs.createWriteStream(destFileName);
await pipeline(readStream, writeStream);
}
copyFile('example.txt', 'copy.txt');
pipeline関数を使用して、読み取りストリームと書き込みストリームを連結します。await pipelineを使用して、パイプライン処理が完了するのを待機します。
async function readStream(fileName) {
const stream = fs.createReadStream(fileName);
for await (const chunk of stream) {
console.log(chunk.toString());
}
await stream.close();
}
readStream('example.txt');
ファイルを別のファイルにコピーする
async function copyFile(sourceFileName, destFileName) {
const readStream = fs.createReadStream(sourceFileName);
const writeStream = fs.createWriteStream(destFileName);
await pipeline(readStream, writeStream);
}
copyFile('example.txt', 'copy.txt');
ファイルの MD5 ハッシュ値を計算する
async function calculateMD5(fileName) {
const stream = fs.createReadStream(fileName);
const hash = crypto.createHash('md5');
for await (const chunk of stream) {
hash.update(chunk);
}
const digest = hash.digest('hex');
console.log(digest);
}
calculateMD5('example.txt');
HTTP リクエストから JSON データを読み込み、パースする
async function fetchAndParseJSON(url) {
const response = await fetch(url);
const stream = response.body.getReader();
let json = '';
for await (const chunk of stream) {
json += chunk.toString();
}
const data = JSON.parse(json);
console.log(data);
}
fetchAndParseJSON('https://jsonplaceholder.typicode.com/posts/1');
- 上記のコードは、Node.js 10.0.0 以降で実行する必要があります。
fsモジュールは、ファイルシステムへのアクセスを提供します。cryptoモジュールは、暗号化機能を提供します。fetch関数は、HTTP リクエストを実行します。
イベントベースの方法は、ストリーム処理の最も伝統的な方法です。この方法では、ストリームからイベントが発行されるたびに、コールバック関数を呼び出します。
const fs = require('fs');
const stream = fs.createReadStream('example.txt');
stream.on('data', (chunk) => {
console.log(chunk.toString());
});
stream.on('end', () => {
console.log('ファイルの読み込みが完了しました。');
});
Promise
Promise を用いた方法は、イベントベースの方法よりも簡潔で、コードを読みやすくすることができます。
const fs = require('fs');
const stream = fs.createReadStream('example.txt');
stream.on('data', (chunk) => {
console.log(chunk.toString());
});
stream.on('end', () => {
console.log('ファイルの読み込みが完了しました。');
});
stream.on('error', (err) => {
console.error(err);
});
Transform streams
Transform streams は、ストリームのデータを変換するために使用することができます。
const fs = require('fs');
const through = require('through2');
const transformStream = through((chunk, encoding, callback) => {
const upperCaseChunk = chunk.toString().toUpperCase();
callback(null, upperCaseChunk);
});
const readStream = fs.createReadStream('example.txt');
const writeStream = fs.createWriteStream('copy.txt');
readStream.pipe(transformStream).pipe(writeStream);
Duplex streams
Duplex streams は、読み込みと書き込みの両方が可能なストリームです。
const net = require('net');
const server = net.createServer((socket) => {
socket.on('data', (data) => {
socket.write(data.toString().toUpperCase());
});
});
server.listen(3000);
javascript node.js async-await
