node.jsのclusterモジュールで、子プロセスからMySQLに接続する方法

Question

node.js v9.11.1
cluster: 0.7.7

node.jsのclusterモジュールを使っているのですが、親プロセスからは問題なくMySQLに接続できるのですが、子プロセスからは繋がりません。

どうやって繋げるのでしょうか？

下の例は親のMySQLの実行結果を子に渡して、子はその値を元にMySQLを実行する、というものです。

子にデータは問題なく渡っているのですが、その先が詰まってしまって困っています。
よろしくお願い致します。

'use strict';

const mysql = require('mysql2/promise');
const cluster = require('cluster');
const CPUs = require('os').cpus().length;

// MySQL接続
const connect = () => {
  const connection = mysql.createConnection({
    host: 'localhost',
    user: 'root',
    password: '',
    database: 'test'
  });
  return connection;
}

// 親プロセス
const master = async () => {
  // MySQL接続
  const conn = await connect().catch((err) => {
    console.log(err);
  });

  // workerプロセスを作成
  for (let i = 0; i < CPUs; i++) {
    const worker = cluster.fork();
  }

  // MySQL実行
  let row;
  const sql = `SELECT id FROM hoge LIMIT 1`;
  try {
    [row] = await conn.query(sql);
  } catch (err) {
    console.log(err);
  }

  // 実行結果を子プロセスに渡す
  try {
    cluster.on('online', (worker) => {
      if (row[0]) worker.send(row[0].id);
    });
  } catch (err) {
    console.log(err);
  }
}

// 子プロセス
const child = async () => {
  // MySQL接続 ←ここが動いていない
  const conn = await connect().catch((err) => {
    console.log(err);
  });

  // 親から値を受け取る
  process.on('message', async (msg) => {
    // MySQL実行
    let row2;
    const sql2 = `SELECT data FROM fuga WHERE id = ? LIMIT 1`;
    try {
      [row2] = await conn.query(sql2, [msg]);
    } catch (err) {
      console.log(err);
    }

    console.log(row2[0].data);
  });
}

if (cluster.isMaster) {
  master();
} else {
  child();
}

Answer 1

２つぐらい問題を見つけました。

###(A)カラム名のtypo

若干弄りながらテストしてる段階で

TypeError [ERR_MISSING_ARGS]: The "message" argument must be specified

とメッセージが出たので、よくよくコードを見たら親プロセスの方の最初のSQLでidカラムを問い合わせているのに、結果セットを参照している箇所がsend(row[0].queue_id)となってることに気づきました・・・
これだとsendにundefinedが渡されることになり「何も指定してない」と見做されこのメッセージが出るんじゃないでしょうか。質問者さんんが実行したときはこのメッセージが出てませんでしたか？

###(B)on('message', ...)
こちらの問題ははっきりこうと言えませんが・・・

(A)を対策して動かしてみると.on('online', ...)が先に動いてその後子供プロセスが動き出しDBへの接続まで進み、その後.on('message', ...)のハンドラーが起動してないように見えました。

一瞬「子供プロセスがmessageハンドラーを登録する前にsendするとメッセージが失われのだろうか？」と思い、

(B1) onlineになってから500msecぐらい時間をおいてsendする

とすると成功したのですが、ハンドラーが登録されてからでないとsendできないは「間違いのような気がした」ので、(B1)の代わりに

(B2) 子供プロセス起動直後にmysqlの接続をせずにmessage受信後に接続する

とするとそちらでも期待通りに動くようでした。なぜ(B2)をやってみたかというと...

mysqlの接続をawaitを用いて接続待ちしている際に親プロセスから既にmessageイベントがキューイングされているはずなのだけどそのハンドラーを登録する前にEmitterを使うような非同期処理をやってしまうとだめなのでは・・・というふうに想像したのです。ちゃんと仕様を調べてないのでこれは単なる勘です。

すみませんが、本当のところはわかってません。どなたか詳しい方がコメントくださると嬉しいのですが・・・（スミマセン）

Comments

[退会済みユーザー]

ありがとうございます。
(A)は単なるケアレスミスだったので修正しました。例ではなくて元のソースの一部が残っていたのでスルーでお願い致します。

[KSwordOfHaste]

(A)の方ー＞了解です。
(B)の方が自分は気持ち悪いです。回答してはみたもののとんでもない間違いを犯している予感がします・・・

Answer 2

解決しました。
cluster.on('online')が今回期待通りの動きをしてくれなかったのがそもそもの原因でした。

自分はclusterモジュールでの親子間の会話が、どちらから始まるのかがわからなかったのですが、今回のケースだと子から始めるとうまくいくとわかりました。

onlineイベントハンドラは単純に子プロセスに繋がれば親に教えてくれるだけなので、子へメッセージを渡してくれません。

そこで子が親にどつかれて何か返事をすれば、親プロセスが子にメッセージを渡すという方法に変えました。

下のようなTABLEがあったとします。

MariaDB [test]> select * from hoge;
+----+-----------------+
| id | fuga            |
+----+-----------------+
|  1 | 福沢諭吉        |
|  2 | 夏目漱石        |
|  3 | 野口英世        |
|  4 | 新渡戸稲造      |
|  5 | 樋口一葉        |
+----+-----------------+

長くなるのでtry catchは省略しました。

'use strict';

const mysql = require('mysql2/promise');
const cluster = require('cluster');
const CPUs = require('os').cpus().length;

// MySQL接続
const mysqlConf = {
  host: 'localhost',
  user: 'root',
  password: '',
  database: 'test',
};

// 親プロセス
const master = (conn) => {
  // 子からメッセージを受け取ったらMySQL接続
  cluster.on('message', async (worker, msg) => {
    if (msg == '__REPLY__') {
      const sql = `SELECT fuga FROM hoge WHERE id = 2`;
      const [row] = await conn.query(sql);
      await worker.send(row[0].fuga); // 子に結果を渡す
    }
  });

  // CPUの数だけフォークする
  for (let i = 0; i < CPUs; i++) {
    cluster.fork();
  };
};

// 子プロセス
const child = async (conn) => {
  // まず子から親に話しかける
  await process.send('__REPLY__');

  // 親から返事を受け取ったらMySQL接続
  process.on('message', async (msg) => {
    const result = `SELECT id, fuga FROM hoge WHERE fuga = ?`;
    const [rowResult] = await conn.query(result, [msg]);
    console.log(rowResult[0].id, rowResult[0].fuga);
  });
};

const main = async () => {
  const conn = await mysql.createConnection(mysqlConf);
  if (cluster.isMaster) master(conn);
  else child(conn);
};

main();

実行結果

2 '夏目漱石'
2 '夏目漱石'
2 '夏目漱石'

上記コードは接続を切っていないので、子プロセスが処理をすべて終えたら接続を切る必要があると思います。

Answer 3

＞(B2) 子供プロセス起動直後にmysqlの接続をせずにmessage受信後に接続する
の方法で一応確認できました。

ただいろいろ試したのですが、私自身も納得のいかない部分もあり、未解決ということにさせて頂きます。

修正した部分は以下です。

const child = async () => {
  // 親から値を受け取る
  process.on('message', async (msg) => {
  // MySQL接続　ここから
  const conn = await connect().catch((err) => {
    console.log(err);
  }); // ここまでをprocess.onの下に移動

    // MySQL実行
    let row2;
    const sql2 = `SELECT data FROM fuga WHERE id = ? LIMIT 1`;
　.
　.
　.
（略）