本連載では分散型マイクロブログ用ソフトウェアMisskeyの開発に関する紹介と、関連するWeb技術について解説を行っています。

今回は、サーバー上で実行されるタスクを管理するシステムであるジョブキューについての全般と、そのライブラリであるBullMQについて紹介します。

ジョブキューとは

ジョブキューは、Webサーバー上で発生する様々なタスク（ジョブ）を管理するシステムで、タスクを複数のサーバーで分散して処理させたり、タスク失敗時にリトライしたりする機能を持ちます。

Misskeyでもジョブキューを活用しており、例えばActivityPubにおけるアクティビティの受信および配送や、ユーザーデータのインポートエクスポート処理をジョブとして管理しています。

Misskeyが使っているジョブキューのライブラリであるBullMQは、毎日指定した時刻にジョブを実行する機能などもあるので、サーバーの統計情報の収集などにも使っています。

ジョブキューの利点

ジョブキューを使う主なメリットは次のとおりです。

• タスクをメインのサーバーから分離して処理できる
  • 処理の負荷が高くなったりしてもサービス全体の動作に影響を及ぼさない
  • 与えられたタスクを処理さえできればいいため、どんなプログラミング言語を使ってもよい
• タスクを複数のサーバーで分散して処理できる
  • データベースに依存しないようなタスクであれば、サーバーを増やせば増やすほど簡単にスケールアウトできる
• 失敗したタスクのリトライを自動で行える
• 使うジョブキューによっては時間指定で起動するタスクや一定間隔で繰り返されるタスクを定義できる
• ダッシュボードが用意されているものもあり、GUIでジョブの状態を把握したり管理ができる

前述したように、Misskeyでは「ActivityPubにおけるアクティビティの受信および配送」がまさにジョブキューにうってつけの処理であるため大いに活用しています。他にもメールの配信といった処理も典型的なジョブキュー向けのタスクでしょう。

このように「⁠『⁠処理にかかる時間や処理が成功するかどうか』が別のサーバーに依存するようなタスク」はジョブキューに向いていると言えます。

BullMQ

Node.jsのジョブキューはいくつかありますが、Misskeyではその中でも BullMQを採用しています。TypeScriptネイティブなこと、活発にメンテナンスされていること、性能が高いこと、定期ジョブを定義できることなどが理由として挙げられます。

他にもBullMQには様々な機能があり、ジョブ間の親子関係を定義できたりもします。

BullMQはバックエンドにRedisを使うようになっています。Misskeyではキャッシュ、Misskeyプロセス間の通信、レートリミットなどにRedisを使っていますが、BullMQ用のRedisを設定できるようになっていたりします。

Bull（無印）との違い

同じジョブキューのライブラリにBullがあります。BullMQはBullのバージョンアップといった位置付けのため、特別な理由がなければBullMQを使いましょう。

Misskeyでも、以前は無印Bullを使用していましたが、最近BullMQに移行する作業を行いました。

使い方

詳細については公式ドキュメントがあるため、ここでは簡単に使い方を紹介します。

BullMQにはQueueとWorkerの２つのクラスが用意されており、Queueがジョブを管理するクラス、Workerが実際にジョブを処理するクラスです。

例としてメールを送信するシステムを考えてみます。まずメール送信用ジョブキューのQueueクラスを作成します。

import { Queue } from 'bullmq';

const emailQueue = new Queue('email');

次にジョブを処理するWorkerクラスを作成します。

import { Worker } from 'bullmq';

const emailWorker = new Worker('email', async job => {
  await sendEmail(job.data.to, job.data.text);
});

• コンストラクタの第一引数には対応するQueueクラスに指定した名前と同じものを設定します（この例ではemail⁠）⁠。
• 第二引数は実際のジョブの処理を行うハンドラで、job.dataにはジョブ追加時に設定されたジョブごとのデータが入ります。今回はメール送信を行うため、送信先アドレスやメール本文などが入ることになります。

最後に、実際にQueueクラスに対してジョブを追加します。

emailQueue.add('alice', { to: '[email protected]', text: 'hi' });
emailQueue.add('bob', { to: '[email protected]', text: 'hello' });

• addの第一引数はジョブ名で、これはデバッグやジョブの種類を判別できるようにするためのラベルのようなものです。
• 第二引数はジョブに持たせるデータで、前述したようにWorkerから参照されます。

これで完了です。あとはジョブキューによってこれらのジョブが処理されます。もし相手のメールサーバーが応答しないなどの理由でジョブが失敗すると、時間を置いて自動で再度ジョブが処理されます。どれくらいの時間を開けるかや、最大で何回リトライするかなどは自由に設定できます。

定期ジョブ

cronのように定期的に実行されるジョブを定義することもできます。

emailQueue.add('alice', { to: '[email protected]', text: 'hi' }, {
	repeat: { pattern: '0 0 * * *' },
});

このようにオプションで繰り返しパターンを定義できます。上記の例では毎日0時0分にメールが送信されます。

仕組み

BullMQ内部では、次のように処理されます。

1. まずジョブが追加されると、Queueによってジョブ情報がRedisに登録される
2. 次に、各WorkerがRedisからジョブ情報を取得し、Workerに登録されているジョブ処理ハンドラを呼び出してジョブの処理を行う
3. 処理が失敗または成功すると、その結果をRedisに書き込む
4. Queue側でそれを把握する

QueueはRedisとやり取りするだけで、Workerについては関知していません。そのためWorkerはいくつでも増やせますし、WorkerがBullMQのものである必要さえありません。BullMQはNodeのライブラリですが、例えばジョブを処理するWorkerだけRustで書くといったことも可能と思います。

冪等性

ジョブキューを利用する上で気をつけたいポイントとして、ジョブの冪等性（idempotency）があります。

冪等性とは「その操作を何回行っても結果に影響しない」という性質です。

BullMQ含めジョブキューは、「⁠そのジョブが最低1回は確実に実行される」ことが保証されていることが多いでしょう。逆に言うと、同じジョブが複数回実行されることもあるということです。これはパフォーマンス上の理由によるものです。

また、それ抜きにしてもジョブは失敗時にリトライされるため、やはりジョブは複数回実行され得ることには変わりありません。

そのため、仮に同じ処理が複数回実行されたとしても問題ないように設計する（＝ジョブの冪等性を保つ）ことが必要です。そうしないとデータの不整合など、思わぬ不具合を引き起こす可能性があります。

例

例えば、ユーザーをフォローするジョブを考えてみましょう。このジョブには、データベースの指定ユーザーのフォロワー数カウントをインクリメントする処理が含まれています。

ここで、単にインクリメントするだけ（＝冪等性が考慮されていないジョブ）だと、このジョブが（同じパラメータで）複数回実行されたときにその分だけインクリメントされ、実際のフォローは1回だけなのにフォロワー数は3増える、といったことが起こりかねません。

このジョブに冪等性を持たせ、このようなことが起こらないようにするには、例えばジョブの最初に「既にフォロー状態であれば後の処理をすべてスキップする」ような処理を入れておくことなどが考えられます。これにより何回このジョブが実行されたとしてもインクリメント処理は1回だけ実行されます。

このように、ジョブを実装する際は「同じ処理が複数回実行され得る」ということを念頭に置いて設計することが大切です。

ジョブの監視

BullMQでは、ジョブ一覧や失敗したジョブについてエラー詳細などを閲覧できるWeb UIが利用できます。

Misskeyではbull-boardを利用していて、サーバー管理者がジョブキューの状態を確認できるようになっています。

まとめ

今回は、サーバー上で処理されるタスクを管理するシステムであるジョブキューの全般的な説明と、Misskeyが使用しているライブラリであるBullMQについて紹介しました。

ジョブキューを利用するにはジョブの設計・実装などひと手間必要ですが、システムの安定性、スケーラビリティ、メンテナンス性を向上できるため、ある程度の規模のサービスにおいては必須と言えます。Amazon SQSなど、SaaSとしても利用可能ですので機会があれば触れてみてください。