FirebaseQueueを利用したWorkerのスケールアウト戦略

はじめに

Advent Calendar初参加のはたです
今回は業務でFirebaseをたまたま利用していたので、丁度いいアウトプットの機会だと思い参加しました！

どのように利用したか

自分不器用なので・・・図にしました。
なんとなくイメージはわかると思います

Node.jsの処理はWebサーバとしての役割ではなく、完全にWorkerとして処理させています。
Workerと言っているのは、タスクキューな処理のことを指しています。
キュー自体はRealtimeDatabaseを活用したものです。
RealtimeDatabaseをQueueにすることで、リアルタイムにTaskの取り出しが行えます。

アプリ側だけでは完結できない処理を、サーバサイド側のWorkerにタスクを依頼する形で
処理を行わせます。性質上RealtimeDatabaseとの相性は抜群だと思いました。

ちなみにこのWorker(Node.js)はRealtimeDatabaseとの通信しか行いません。
他にも連携しているサービスはありますが、今回はFirebase周りに絞りました。
ご覧の通り、Googleさんにお世話になっております。

Firebaseとjs

構築した最初のWorkerは下記のよう形になっていました。

RealtimeDatabase

{
  "queue" : {
    "queue_id_1" : {
      "type" : 1,
      "data": {}
    },
    "queue_id_2" : {
      "type" : 1,
      "data": {}
    },
    "queue_id_3" : {
      "type" : 2,
      "data": {}
    }
  }
}

js

db.child('/queue').on("child_added", function(snapshot, prevChildKey) {
  var task = snapshot.val();
  // task.typeの値によって適切な処理を行う
  // 処理が終わったらtaskを削除
});

/queueに追加されるデータを監視して、taskを処理するという形です。
これをベースにすれば、Workerの役割を実現できそうでした。

スケールアウト問題

当然ユーザ数の増加に伴う処理の負荷対策を考えなければいけません。
今回は単体の性能をアップさせるスケールアップより、サーバ自体の台数を増やすスケールアウトの戦略で負荷への対策を考えました。
しかし上記で紹介した方法だととある問題が・・・

問題点

紹介したjsのように、単純にchild_addedをするだけですと
リアルタイム通信している２台のworkerが同じタスクを処理してしまうということが起きます。
同じタスクをworkerが取り出さないように工夫が必要になりますね。
自分でやろうとすると少々面倒です・・・

と、悩んでいる時に見つけたのが
Firebase Queue

Firebase Queue

https://github.com/firebase/firebase-queue
https://github.com/firebase/firebase-queue/blob/master/docs/guide.md

Firebaseの公式がいくつかライブラリを提供していますが、そのうちの１つになります。
このライブラリの説明は

A fault-tolerant, multi-worker, multi-stage job pipeline built on the Firebase Realtime Database.

なんと！！
悩んでいたことはこいつがなんとかしてくれそうです！

Firebase Queueを使ってみた

準備

npm install firebase firebase-queue --save

データ構造

{
  "queue" : {
    "tasks": {
    }
  }
}

当然rulesも必要ですが、今回は省きます

実装するjs

Worker側

const TASK_PUSH = 1;

var Queue = require('firebase-queue');
var admin = require('firebase-admin');

var serviceAccount = require('path/to/serviceAccountCredentials.json');
admin.initializeApp({
  credential: admin.credential.cert(serviceAccount),
  databaseURL: '<your-database-url>'
});

var ref = admin.database().ref('queue');
var queue = new Queue(ref, function(data, progress, resolve, reject) {
  switch (data.type) {
    case TASK_PUSH:
      push(data.value);
      resolve();
  }
});

実際にはqueue/tasks/のデータを監視してくれます。
resolve()することによってqueue/tasks/からデータを削除してくれます。
つまり処理が完了した場合にresolve()すればいいわけです。
(errorの場合、rejectするとデータは残ります。）

optionsとか

var options = {
  'specId': 'spec_1',
  'numWorkers': 5,
  'sanitize': false,
  'suppressStack': true
};
var queue = new Queue(ref, options, function(data, progress, resolve, reject) {
  ...
});

こんな感じにoptionが指定できますが、
numWorkersなんかは、1プロセスのNode.jsで、RealtimeDatabaseとの通信を行う数を指定できたりします。

余談

ちなみにnumWorkersの値分、RealtimeDatabaseとの通信が発生、つまりWebSocketのコネクションをはります。このコネクションでタスクの取り出しを行うわけですが、取り出しを行ったあとresolveが呼ばれるまでこのコネクションの状態をロジック的にbusyとして(busyフラグを立てる）新規取り出しの処理をブロックします。
つまりresolveしないで放置しておくと、すべてのコネクションが新規取り出しをブロックして、
Queueの処理がうんともすんとも言わなくなります。
なのでタスクの処理が終わる or 終わると同等のタイミングになったらresolveはちゃんと呼びましょう。(rejectも同じ）

タスク登録側

var ref = firebase.database().ref('queue/tasks');
ref.push({'type': 1, 'value': true});

(jsの紹介ですが、本来はアプリから登録してもらいます

データの動き

{
  "queue" : {
    "tasks": {
      "xxxxx": {
        "_owner": "xxxxxxxxxxxx",
        "_progress" : 0,
        "_state" : "in_progress",
        "_state_changed" : 1478484860167,
        "_id": "xxxxxx",
        "type" : 1,
        "value" : true
      }
    }
  }
}

queue/tasksにデータを登録するとFirebaseQueueが自動で必要な値を付け足します。(_のやつ)
この値によって、タスクの取り出しに重複がないようライブラリ側で色々制御しています。
(xxxxはユニーク値です)

Firebase Queueのいいところ

• タスクの処理が重複しないよう制御されている
  (あるTaskがキューから取り出されたら、そのTaskが他の通信によって取り出されることはないよう保証されている）
• データ構造がPromiseのため、resolveやrejectを利用できる
• specsを利用すれば処理のチェーンなども実現できる
• 公式の安心感

おわりに

実際に実装したWorkerのことに合わせて、FirebaseQueueの基本的な紹介をしました。
ドキュメントを読み込んでみると、色々応用ができそうですね。

Firebaseをフル活用するために、ちゃんと公式からGoodなライブラリが出てることには感動しました。
ありがとうFirebase、ありがとうFirebaseQueue。

皆さまもぜひ、Firebaseだけではなくその周辺ライブラリも見てみてください。
困った時に意外と役立つライブラリや情報があるかも！？

それでは良いFirebaseライフを。