Node.jsのマルチスレッド入門シリーズ：worker_threadsチュートリアル

Node.jsはシングルスレッドのJavaScript実行環境を提供し、コードが複数の処理を並列に実行することを防ぎます。アプリケーションが通常、同期実行を行う場合、時間のかかる処理中に処理がブロックされることがあります。

ただし、Node.js自体はマルチスレッドアプリケーションです。これは、ファイルの読み込みやネットワークリクエストの送信など、標準ライブラリの非同期メソッドを使ってI/O処理を行うと分かります。こうしたタスクは、Node.jsがCライブラリのlibuvを使って作成・管理する別のスレッドプールに委譲されます。マルチスレッドのように見えますが、非同期関数がメインのイベントループをブロックする可能性は依然としてあります。

では、Node.jsアプリケーションで独立したスレッドを自分で作成したい場合はどうでしょうか。マルチスレッドを使うと、任意の処理を並列に実行できるため、CPU負荷の高いワークフローのパフォーマンスを大幅に向上できます。Node.jsは真のマルチスレッドを提供していませんが、worker_threadsモジュールを使えば、同様の仕組みを実現できます。この記事では、その機能と、実際のユースケースでの使い方を解説します。

worker threadsとは？

worker_threadsモジュールは、アプリケーションに並列処理を追加できるスレッド機能を実装します。worker threadで実行するコードは独立した子プロセスで動作するため、メインアプリケーションをブロックしません。

worker threadsは、従来の意味での真のスレッドではありません。独立したプロセスであるため、親プロセスの実行コンテキストに直接アクセスすることはできません。worker threadsとアプリケーション間の通信には、イベントベースのメッセージングシステムが使われます。

worker threadsによってNode.jsが真のマルチスレッド言語になるわけではありませんが、多くの実際のシナリオでは、その違いは実質的な問題になりません。複数の実行「スレッド」を同時に動かす便利な仕組みとして、負荷の高い処理をメインループから切り離せます。

worker threadsのユースケース

worker threadsは、CPU負荷の高い処理であれば幅広く活用できます。ただし、スレッドごとにオーバーヘッドが発生するため、I/O処理の高速化には適していません。ファイルシステムやネットワークのタスクには、Node.jsに組み込まれた非同期I/Oユーティリティの方が高速で効率的です。

worker threadsが役立つ場面は数多くあります。特に一般的で、導入による効果が期待できるユースケースをいくつか紹介します。

• 画像のリサイズ：大きな画像のリサイズには数秒かかることがあり、複数のサイズを生成する場合は遅延がすぐに積み重なります。アップロードされた写真をサムネイルや、小・大サイズの画像に変換するアプリケーションでよくあるケースです。3つのworker threadsを使ってすべてのサイズの生成を同時に開始すれば、処理全体の時間を短縮できます。

• 動画の圧縮：動画の圧縮は、最も計算負荷の高い処理の1つです。worker threadsを使って複数のフレームを並列に処理し、結果をメインスレッドに返すことで高速化できます。

• ファイル暗号化などの暗号処理：暗号処理は意図的に複雑に設計されています。ファイルの暗号化と復号、秘密鍵の生成、署名の検証などをメインスレッドで行うと、プログラムに目立った遅延が生じることがあります。

• 大量のデータの並べ替えと検索：データのフィルタリングやソートでは、各値を比較するために大量の反復処理が必要です。worker threadsを使って複数のデータを並列に処理すれば、高速化できます。

• 複雑な数学的処理：素数の生成、大きな数の素因数分解、複雑なデータ分析などの数学的計算は、本質的にCPU負荷の高い処理です。一部の処理を別スレッドで実行すれば、メインループを他のタスクに使えるようになります。

これらすべての処理が遅くなるのは、ディスクやネットワークからデータを読み込むのではなく、CPUがコードの実行に多くの時間を費やすためです。反復処理であるため、並列に実行する処理の回数を増やすことが、パフォーマンス向上への最善の方法です。worker threadsは、それを実現する仕組みです。

worker threadsモジュールの使い方

Node.jsのworker_threadsモジュールをインポートし、new Worker()コンストラクターを呼び出すと、worker threadを作成できます。指定したJavaScriptファイルを実行する新しいプロセスが起動します。workerとの間でメッセージをやり取りし、イベントの発生、処理対象データの準備完了、エラーの発生などを確認できます。ブラウザーでWeb Workersを使ったことがあれば、worker threadの概念にもなじみがあるでしょう。

JavaScriptファイルをworker threadで実行する、最も簡単な例を見てみましょう。

1const {
2    Worker,
3    isMainThread,
4    parentPort
5    workerData
6} = require("worker_threads");
7
8if (isMainThread) {
9    const worker = new Worker(__filename, {workerData: "hello"});
10    worker.on("message", msg => console.log(`Worker message received: ${msg}`));
11    worker.on("error", err => console.error(error));
12    worker.on("exit", code => console.log(`Worker exited with code ${code}.`));
13}
14else {
15    const data = workerData;
16    parentPort.postMessage(`You said \"${data}\".`);
17}

このコードスニペットをコピーし、作業ディレクトリにworker-demo.jsという名前で保存します。Node.jsで実行すると、次のような出力が表示されます。

1$ node worker-demo.js
2Worker message received: You said "hello".
3Worker exited with code 0.

このシンプルなコードでは、worker threadsの基本的な仕組みをすべて確認できます。ソースコードは、メインスレッドとしてもworker threadとしても動作します。worker_threadsモジュールにはisMainThreadというエクスポートがあり、コードがメインスレッドで実行されているかどうかを確認できます。

ファイルがメインスレッドとして実行されている場合、つまりターミナルから起動した場合は、if文の最初の分岐が実行されます。ここで新しいWorkerインスタンスを作成します。

Workerコンストラクターの最初の引数には、workerで実行するJavaScriptファイルのパスを指定します。この例では、同じファイルにメインスレッドとworker threadの両方のコードが含まれているため、グローバル変数__filenameを使えます。Workerの2つ目の引数にはオプションオブジェクトを指定します。このオブジェクトのworkerDataプロパティを使うと、worker threadに値を渡せます。

メインスレッド部分の残りのコードでは、Workerインスタンスにイベントリスナーを設定します。これにより、worker内で発生したメッセージ、エラー、プロセスの終了に対応できます。

サンプルコードがworker threadとして実行されると、isMainThread変数はfalseになり、elseの分岐が実行されます。この分岐では、worker_threadsモジュールのworkerDataエクスポートを使って、メインスレッドから渡されたデータにアクセスします。parentPortエクスポートはメインスレッドとのインターフェースを提供し、postMessage()関数を使うとメインスレッドにデータを送り返せます。データはmessageイベントとして通知されます。

つまり、worker-demo.jsを実行すると、次の順番で処理が行われます。

• 新しいworkerが作成され、データとしてhelloが渡されます。

• worker内で実行されるコードがデータを取得し、新しい親しみやすいメッセージをメインスレッドに返します。

• メインスレッドのイベントリスナーがworker threadから送られたメッセージを受け取り、コンソールに出力します。

• worker threadで実行するコードがなくなると、workerは終了します。メインスレッドにはexitイベントで通知されます。

それでは、実際のユースケースでworker threadを使ってみましょう。

worker threadsを使った画像のリサイズ

このコードでは、画像を3種類のサイズに並列で変換します。各サイズを順番に処理するよりも、はるかに高速です。

まず、画像をリサイズするコードを作成します。resize-worker.jsという名前で保存してください。

1const {parentPort, workerData} = require("worker_threads");
2const sharp = require("sharp");
3
4const {src, width, height} = workerData;
5const [filename, ext] = src.split(".");
6
7console.log(`Resizing ${src} to ${width}px wide`);
8
9const resize = async () => {
10await sharp(src)
11.resize(width, height, {fit: "cover"})
12.toFile(`${src}-${width}.${ext}`);
13};
14
15resize();

新しいファイルにメインスレッドのコードを作成し、resize-main.jsという名前で保存してください。

1const {Worker} = require("worker_threads");
2
3const src = process.argv[2];
4
5const sizes = [
6{width: 1920},
7{width: 1280},
8{width: 640}
9];
10
11for (const size of sizes) {
12const worker = new Worker(
13    __dirname + "/resize-worker.js",
14    {
15        workerData: {
16            src,
17            ...size
18        }
19    }
20);
21}

画像のリサイズ機能を提供するsharpモジュールをnpmでインストールします。

1$ npm install sharp

次に、大きな画像を作業ディレクトリに置き、image.jpgという名前を付けます。Unsplashにある色鮮やかな糸の写真を使ってもよいでしょう。次のコマンドでコードを実行します。

1$ node resize-main.js image.jpg

「Resizing」というメッセージが3つ、すぐに表示されます。メインスレッドは指定されたサイズを順に処理し、それぞれに新しいスレッドを作成して、処理の完了を待ちます。こうすることでメインスレッドをブロックせずに、サイズを並列で生成できます。

worker threadsを使った動画のリサイズ

このコードスニペットでは、[FFmpegを使ってworker threadで動画をリサイズする方法を紹介します。負荷の高い動画のリサイズを実行しながら、メインスレッドで他のタスクを続けられます。

workerのコードを新しいファイルに保存し、video-worker.jsという名前を付けます。

1const {parentPort} = require("worker_threads");
2const ffmpeg = require("fluent-ffmpeg");
3
4const resizeVideo = (src, size) => {
5    const [filename, ext] = src.split(".");
6    const output = `${__dirname}/${filename}-${size}.${ext}`;
7    ffmpeg(`${__dirname}/${src}`)
8    .size(size)
9    .on("end", () => parentPort.postMessage({output, input: src, type: "done"}))
10    .save(output);
11};
12
13parentPort.on("message", msg => {
14    const {file, size} = msg;
15    const [filename, ext] = file.split(".")[0];
16    resizeVideo(file, size);
17});

次に、メインスレッドのコードを新しいファイルに保存し、video-main.jsという名前を付けます。

1const {StaticPool} = require("node-worker-threads-pool");
2
3const pool = new StaticPool({
4    size: 4,
5    task: __dirname + "/video-worker.js"
6});
7
8const videoToResize = process.argv[2];
9const videoTargetSize = process.argv[3];
10
11const resize = async () => {
12    await pool.exec({file: videoToResize, size: videoTargetSize}).then(() => {
13    if (msg?.type === "done") {
14    console.log(`Saved ${videoToResize} to ${msg.output}`);
15    }
16    });
17};
18
19resize();

コードに必要なnpm依存関係をインストールします。

1$ npm install fluent-ffmpeg node-worker-threads-pool

この例では、広く使われているffmpegエンコーダーを利用します。fluent-ffmpegパッケージは、システムにすでにインストールされているffmpegライブラリをNode.jsから使うためのラッパーです。コードを実行する前に、[ffmpegをインストールしてください。ほとんどのLinux環境にはすでに含まれていますが、WindowsやMacでは手動で入手する必要がある場合があります。

次のコマンドを実行すると、ffmpegが利用可能かどうか確認できます。

1$ ffmpeg
2ffmpeg version 4.2.7-0ubuntu0.1 Copyright (c) 2000-2022 the FFmpeg developers

バージョン番号が表示されれば、すでにインストールされています。「command ‘ffmpeg’ not found」のようなメッセージが表示された場合は、ffmpegのウェブサイトの案内に従って、お使いのシステムに適したライブラリをダウンロードしてください。ffmpegは、主要なLinuxディストリビューションのほとんどのパッケージリポジトリと、macOSのHomebrewからも入手できます。

1$ brew install ffmpeg

これで、動画ファイルをバックグラウンドでリサイズできます。手元にファイルがない場合は、Pexelsにあるこの動画をダウンロードしてください。作業ディレクトリにvideo.mp4という名前で保存します。

次に、スクリプトを実行して動画を1280x720に縮小します。

1$ node video-main.js video.mp4 1280x720
2Saved video.mp4 to /path-to-video-1280x720.mp4

この例は、前の例とは少し仕組みが異なります。効率を高めるため、「ワーカープール」という考え方を使います。前の例とは異なり、リサイズが完了してもプロセスは自動的に終了しません。worker threadは引き続き動作してメッセージを待ち受けるため、ターミナルでCtrl+Cを押して終了する必要があります。

ワーカープールを使うと、リソース消費を抑えられます。workerを作成するたびにオーバーヘッドが発生するため、決まった数のスレッドを起動して、可能な限り再利用するのが効果的です。Poolifier、Piscinaなどのライブラリを使えば、この方法でworker threadsを簡単に利用できます。これらのライブラリは、ユーザーが指定した上限までスレッドを作成し、新しいタスクを処理します。追加のタスクは、既存のスレッドが利用可能になり次第、割り当てられます。

workerを使ったファイルの暗号化

この例では、worker_threadsとNode.jsのcryptoモジュールを組み合わせ、別スレッドでファイルを暗号化する方法を紹介します。

まず、worker threadのコードをencrypt-worker.jsという新しいファイルに保存します。

1const {parentPort, workerData} = require("worker_threads");
2const crypto = require("crypto");
3const fs = require("fs");
4
5const {file} = workerData;
6const output = `${file}.encrypted`;
7
8const key = crypto.randomBytes(32);
9const iv = crypto.randomBytes(16);
10
11const cipher = crypto.createCipheriv("aes-256-ctr", Buffer.from(key), iv);
12const readStream = fs.createReadStream(file);
13const writeStream = fs.createWriteStream(output);
14readStream.pipe(cipher).pipe(writeStream);
15writeStream.on("close", () => parentPort.postMessage({key: key.toString("hex"), output, type: "done"}));

このメインスレッドのコードをコピーし、encrypt-main.jsという新しいファイルに保存します。

1const {Worker} = require("worker_threads");
2
3// Perform some other tasks here
4
5const fileToEncrypt = process.argv[2];
6
7const worker = new Worker(
8    __dirname + "/encrypt-worker.js",
9    {
10    workerData: {
11    file: fileToEncrypt
12    }
13    }
14);
15
16worker.on("message", msg => {
17    if (msg?.type === "done") {
18    console.log(`File encrypted to ${msg.output}`);
19    console.log(`The key is ${msg.key} - don't lose it!`);
20    }
21});
22
23// Perform some other tasks that don't need to wait for the encryption here

次のコマンドをターミナルで実行し、暗号化するテキストファイルを作業ディレクトリに作成します。

1$ echo foobar > demo.txt

次に、コードを実行してファイルを暗号化します。

1$ node encrypt-main.js demo.txt
2File encrypted to demo.txt.encrypted
3The key is 20eb1974c95553ff4f16638192da8b4cbfd780cec8544579f04cde181b78bd7c - don't lose it!

この例では、worker threadからメッセージを送信し、出力ファイルのパスと生成された秘密鍵をユーザーに知らせます。

worker threadsの基礎をおさらい

Node.jsはシングルスレッドのランタイムです。これは、JavaScriptが同期的に実行をブロックする言語であり、すべてを1つのスレッドで実行することに由来します。Node.jsの非同期標準ライブラリコンポーネントがノンブロッキングに見えるのは、別のスレッドプールを使うためです。

worker_threadsモジュールを使うと、同様の仕組みを独自のコードに組み込めます。真のマルチスレッドではありませんが、多くの場面では十分に近い動作を実現し、メインスレッドの外でコードを並列実行できます。

この記事では、ワーカースレッドとは何か、どのような場合に使用できるのか、そしてプロジェクトで使い始める方法について説明しました。このシリーズの次回の記事では、ワーカースレッドのメリットとデメリットを掘り下げ、他のプログラミング言語におけるマルチスレッドの実装と比較します。