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

前回、[Node.js: CPU負荷で3秒かかっていた処理を「Worker Threads」で1秒に時短する](https://qiita.com/suin/items/bce351c812603d413841)という投稿をしました。

本稿では、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`を実行してみると、ワーカーがデータを受信できていることがわかります:

```bash
$ 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つのワーカーでデータの送受信ができているのがわかります:

```bash
$ 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`を通じて行う。