Node.js Worker Threads: スレッド間でデータを送受信する方法

前回、Node.js: CPU負荷で3秒かかっていた処理を「Worker Threads」で1秒に時短するという投稿をしました。

本稿では、Node.jsで本物のスレッドが扱えるWorker Threadsにて、スレッド間でデータを送受信する方法について解説します。

本稿で分かること

• メインスレッドからワーカーをデータを送信するのはどうやるのか？
• ワーカーからメインスレッドにデータを送信するのはどうやるか？
• ワーカー同士のデータの送受信は？

メインスレッドからワーカーへデータを送信する方法

メインスレッドからワーカーへデータを送信するには、WorkerのpostMessageメソッドを用います。次の例は、'Hello!'という文字列データをワーカーに送信するものです:

main.js

const {Worker} = require('worker_threads')
const worker = new Worker('./worker.js')
worker.postMessage('Hello!')

ワーカーでこのデータを受信するには、worker_threadsモジュールのparentPortを使います。この、parentPortオブジェクトには、onメソッドが生えており、'message'イベントを処理するイベントハンドラーを登録することで、メインスレッドからのデータを受信できるようになります:

worker.js

const {parentPort} = require('worker_threads')
parentPort.on('message', message => {
  console.log('worker received message: %o', message)
})

このサンプルコードのmain.jsを実行してみると、ワーカーがデータを受信できていることがわかります:

$ node main.js
worker received message: 'Hello!'

ワーカーからメインスレッドにデータを送信する方法

逆に、ワーカーからメインスレッドにデータを送信するには、parentPortオブジェクトのpostMessageメソッドを使用します。次のサンプルコードは、ワーカー側のコードで、メインスレッドに文字列の'Hello!'を送信するものです:

worker.js

const {parentPort} = require('worker_threads')
parentPort.postMessage('Hello!')

受信側のメインスレッドのコードでは、生成したWorkerオブジェクトのonメソッドにて、'message'イベントを処理するイベントハンドラを登録しておくことで、ワーカーから送信されたデータを受信できます:

main.js

const {Worker} = require('worker_threads')
const worker = new Worker('./worker.js')
worker.on('message', message => {
  console.log('Main thread received message: %o', message)
})

ワーカー同士のデータの送受信方法

ワーカーAから、別のワーカーBにデータを送信するには、メインスレッドが送受信を中継してあげる必要があります。

まず、データ発信者のワーカーは、メインスレッドにデータを送信するようにします:

worker1.js

const {parentPort} = require('worker_threads')
parentPort.postMessage('message from worker1')

次に、データ受信者のワーカーは、メインスレッドからデータを受信するようにします:

worker2.js

const {parentPort} = require('worker_threads')
parentPort.on('message', message => {
  console.log('worker2 received message: %o', message)
})

最後に、2つのワーカーの送受信を中継するコードをメインスレッドに実装します:

main.js

const {Worker} = require('worker_threads')
const worker1 = new Worker('./worker1.js')
const worker2 = new Worker('./worker2.js')
worker1.on('message', message => worker2.postMessage(message))

このmain.jsを実行すると、2つのワーカーでデータの送受信ができているのがわかります:

$ node main.js
worker2 received message: 'message from worker1'

MessageChannelを使ったワーカー間データ送受信

上の例では、メインスレッドでメッセージを中継する方法で、ワーカー間のデータ送受信を実現する方法を紹介しました。

ワーカー間でデータ送受信をする別の方法として、MessageChannelを使う手段があります。

メイン側でMessageChannelを作ると、MessagePortが2つ生成されます。それぞれを、各ワーカーに渡すようにします。

main.js

const {Worker, MessageChannel} = require('worker_threads')
const {port1, port2} = new MessageChannel()
const worker1 = new Worker('./worker1.js')
const worker2 = new Worker('./worker2.js')
worker1.postMessage({worker2: port1}, [port1])
worker2.postMessage({worker1: port2}, [port2])

MessagePortを受け取ったワーカーは、以後MessagePortを通じてワーカー間データ送受信ができるようになります。

worker1.js

const assert = require('assert')
const {parentPort, MessagePort} = require('worker_threads')
parentPort.once('message', ({worker2}) => {
  assert(worker2 instanceof MessagePort)
  worker2.postMessage('message from worker1')
})

worker2.js

const assert = require('assert')
const {parentPort, MessagePort} = require('worker_threads')
parentPort.once('message', ({worker1}) => {
  assert(worker1 instanceof MessagePort)
  worker1.on('message', message => {
    console.log('worker2 received message: %o', message)
  })
})

まとめ

• メインスレッドからワーカーをデータを送信するのは、メインスレッド側でworker.postMessageを呼び出し、ワーカー側は、parentPort.on('message', ...)でイベントハンドリングする。
• ワーカーからメインスレッドにデータを送信するのは、ワーカー側でparentPort.postMessageを呼び出し、メインスレッド側は、worker.on('message', ...)でイベントハンドリングする。
• ワーカー同士のデータの送受信には、メインスレッドによる中継、もしくは、MessageChannelを通じて行う。