MySQLでサーバ間ロック/スケジューラ/キュー

最近は何を作るにしてもマイクロサービスで行くことが多くなってきました。マイクロサービスの中心にはイベントバスとして何らかのキューがあるのが主流だと思いますが、MySQLやREDISに加えてキューのミドルウェアやスケジューラまで入ってくるとコード量はコンパクトでもシステム全体としてみると部品点数の多い巨大なサービスになりがちです。

既存の基盤システムに載せるのならそれでもいいんですが、独立したシステムだとなるべくコンパクトに作りたいですよね。そこで、キューやスケジューラには専用のミドルウェアが必ずしも必要ではなく、それMySQLでできるよということを紹介しておきたいと思います。なお、ただのSQL芸なのでMySQLによらず他のRDBでもできます。

プロセス間ロック

まずはプロセス・サーバー間のロック機構からです。ロックを単体で使う必要性はあんまり思いつかないのですが、技術的にはスケジューラやキューを実現する上での基礎となります。
と言ってもMySQLでマルチクライアントで1つのレコードの処理を排他で行うという処理であれば、当たり前の技術として実践していることが多いと思います。まずは最も単純なロック機構を考えてみましょう。

-- 作成
CREATE TABLE lock_table (
  lock BOOLEAN NOT NULL DEFAULT FALSE
);
INSERT lock_table VALUES(FALSE); -- 1レコード作成
-- ロック取得
UPDATE lock_table SET lock = TRUE WHERE lock = FALSE;
-- アンロック
UPDATE lock_table SET lock = FALSE;

-- ※ちなみに、MySQLのBOOLEANはTINYINT、FALSEは0、TRUEは1のエイリアスです

UPDATEで1件更新されればロックを取得できたことになります。UPDATEできなければ誰かがロックしているのでしょう。その場合はたとえば1分単位でポーリングしてロック解放を待ちます。

ロックに必要なのはこれだけです。ロック操作にはアトミック性が求められますが、1つの文はアトミックに実行されるので、トランザクションも不要です。
しかし、このロックには次のような問題があります。

• どのプロセスがロックを持っているのか（人間に）わからずデバッグが難しい
  • また、できればロック取得をUPDATEの戻り値ではなくSELECTで知りたい
• ロックを持ったままプロセスが死ぬと永久に解放されない

一つ目の問題はプロセスにユニークなIDを持たせて、それをテーブルに書くようにすれば解決します。IDとしてはサーバ名やIPアドレスなど、サーバを特定するものと、サーバ内で一意であることを示すプログラム名やPIDが使えそうです。サーバ内でそのテーブルを使うのが必ず1プロセスであることを別の手段(たとえばsupervisordで起動する)で確約できれば、サーバ名だけでも十分でしょう。

二つ目の問題は、ロックにロックの制限時間を持たせることで解決します。ロック制限時間の持たせ方は複数思いつきます。

• ロック開始時刻と期間
• ロック開始時刻のみ（固定期間）
• ロック終了時刻のみ

ここで最適だと考えられるのは「ロック終了時刻のみ」です。ロック開始時刻の方が、①いつロックを必要にしたのかもわかり、②ロック終了時刻を計算しなくてもいい気がするので、魅力があるように見えるかもしれません。しかし、①の理由は本質的な機能ではなくデバッグログですし、②はそのロックが再利用可能かどうか考えるときには結局ロック終了時刻を考える必要があるので、難しさは変わりません。ログとしては有用なのでロック終了時刻に加えてロック開始時刻を記載しておくのは、良いことだと思います。
また、ほとんどの場合は、ロックは1つの用途にしか使われないはずですので、固定期間で良いように見えるかもしれませんが、データ構造のみで完結していないので、固定期間がバージョンアップなどで変更になった場合に思わぬ事故を引き起こします。
ロック終了時刻をカラムとすることで、比較条件が必要以上に複雑にならず、大規模化した場合にインデックスも効かせられるメリットもあります。

改良ロック

-- 作成
CREATE TABLE lock_table (
  lock_by VARCHAR(32), -- プロセスユニーク名
  expire_at TIMESTAMP NOT NULL -- ロック解放時間
);
INSERT lock_table VALUES(null, '1970/01/01');
-- ロック取得(最大5分間)
UPDATE lock_table SET lock_by = 'HOSTNAME', expire_at = NOW() + INTERVAL 5 MINUTES WHERE expire_at < NOW();
SELECT * FROM lock_table WHERE lock_by = 'HOSTNAME';
-- アンロック
UPDATE lock_table SET expire_at = NOW() - INTERVAL 1 SECONDS WHERE lock_by = 'HOSTNAME';
-- ロックリミット更新
UPDATE lock_table SET expire_at = NOW() + INTERVAL 5 MINUTES WHERE lock_by = 'HOSTNAME';

というわけで、こうなります。
lock_byには前述のとおり、サーバ名やIPアドレスなどを入れておきます。必要に応じて正規化してもいいと思います。expire_atはロックの解放予定時間です。この例ではロック時に (現在時刻+5分) の値を入れています。5分を過ぎると自動的にロックが解放されます。これでサーバやプロセスの突然死・フリーズ時でも安心です。
ロックを長く持ちたい場合は単にexpire_atを長く設定するか、定期的にexpire_atの更新を行います。突然死に対する他プロセス・他サーバによるリカバリはロック時間が短い方が有利ですが、あまりに短いと処理時間が何らかの都合で少し伸びただけでロックを失う可能性があります。
また、定期的なロック時間更新を行う場合、システムが正しく処理を行っている場合のみ更新できるように注意深くコードを書かなければなりません。たとえば、メインスレッドはバグで無限ループに入ってしまい処理が進んでいないのに、5分ごとにタイマースレッドで問答無用で更新しているために、無限にロックされていた、というようなことが起こり得ます。どちらがマシかをアプリケーションごとに判断して適切な動作を選びます。
例では処理時間が伸びて万が一の場合にロックを途中で失った場合に、より不可解な現象を起こさないように、アンロック・リミット更新時にはWHERE句でlock_byを調べています。本質的には必要ありませんが、ほぼノーコストな保険なので、付けておいた方が良いでしょう。

複数のロック

このlock_tableはただのテーブルなので、resource_nameなどを入れて複数レコードを用意することでリソースプールを作ることもできます。lock_tableに対するアトミック性を失わなければなんでもできます。

状態の記録

expire_atだけで解放かロックかを表すと、解放時にexpire_atの値を操作する必要があって、スマートではありません。expire_atを「（未来の）いつ強制解放されるか」と捉えると、解放時に「（過去の）いつ正常解放されたか」という意味合いに化けているので、少し収まりが悪いですね。番人と考えて、そこで'1970/01/01'を投入してもいいのですが、レコードだけを見たときには若干意味不明になりますし、かといってNULLを使ってしまうとWHERE句が妙に複雑化しそうです。
そこで、少なくともロックしているかどうかのフラグまたはenumを別カラムで用意すると、少しの複雑性で解放時にexpire_atを操作する必要が無くなり、最後にロックがきちんと解放されたのか、単にロックがタイムアウトしたのかを示すこともできるようにもなります。

-- 作成
CREATE TABLE lock_table (
  lock_by VARCHAR(32), -- プロセスユニーク名
  state ENUM ('FREE', 'LOCK', 'EXTEND') NOT NULL,
  expire_at TIMESTAMP NOT NULL -- ロック解放時間
);
INSERT lock_table VALUES(null, 'FREE', '1970/01/01');
-- ロック取得(最大5分間)
UPDATE lock_table SET lock_by = 'HOSTNAME', state = 'LOCK', expire_at = NOW() + INTERVAL 5 MINUTES WHERE expire_at < NOW() OR state = 'FREE';
SELECT * FROM lock_table WHERE lock_by = 'HOSTNAME';
-- アンロック
UPDATE lock_table SET state = 'FREE';
-- ロックリミット更新(このstate更新は本質的には無意味。ただのデバッグログ)
UPDATE lock_table SET state = 'EXTEND', expire_at = NOW() + INTERVAL 5 MINUTES WHERE lock_by = 'HOSTNAME';

実践時には定番のid/insert_time/update_timeの他にもこういった情報を含めた方がデバッグ性が向上するかと思います。

lock_byについてもう少し

lock_byを利用してUPDATEでマーキングしてからSELECTするようにしましたが、別の方法として、UPDATEで条件を満たしたレコードをマーキングする代わりに条件を満たしたレコードをSELECTして、ID指定でUPDATEして、UPDATEできた場合にロックできたと考える（またはトランザクション処理で確実に行う）とすることもできます。しかし、そのやり方はクエリ発行量やテーブルロックが増えるだけでそれほど機械的なメリットは無いように思います。（SQLの美しさやわかりやすさについては議論があるかも知れません）
スケジューラやキューでもマーキング用に同じ役目のカラムが出てきますが、同意です。

スケジューラ

次に複数のサーバ間で協調して動くCRONのようなものを考えます。時間が来たら冗長化構成のいずれか1台のみがコマンドを実行できるものです。
最も簡単に考えると、タスクをどのタイミングで実行すべきかをDBに入れておいて、実行時にタスクレコードをロックすることで1台だけの実行を担保する実装が思い浮かびます。しかし、きちんと考えると一筋縄ではいきません。実行予定時刻にサーバAが処理を実行しているときにサーバBも実行しようとした場合は正しく動きます。しかし、その処理実行が一瞬で終わり、サーバAがアンロックした後でサーバBが実行しようとした場合にはサーバBが再度そのタスクを実行することを許してしまいます。したがって、最後にいつ実行したかをきちんと記録して照らし合わせなければなりません。

-- 作成
CREATE TABLE cron_table (
  id INTEGER NOT NULL,
  hour INTEGER NOT NULL,
  command VARCHAR(255) NOT NULL,
  last_exec TIMESTAMP
);
-- タスク登録
INSERT lock_table VALUES(1, 4, 'echo 4時になったよ', '1970/01/01');
-- タスク取得(日次で4時に実行した場合)
SELECT * FROM lock_table WHERE hour = 4;
-- タスク実行権限取得(2015年12月13日4時に実行した場合)
UPDATE cron_table SET last_exec = '2015/12/13 04:00' WHERE id = 1 AND last_exec < '2015/12/13 04:00';

1日の中でn時ちょうどになったらcommandを開始するスケジューラの簡単な例です。ワーカーは少なくとも毎時間ごとに各サーバでCRONなどで起動する必要があります。実際にはsupervisordなどでデーモン化するのがベストだと思います。
面倒なのは、last_execの更新と比較です。last_execは実行間隔による境界値と比較する必要があります。たとえば、4時に実行しようとした場合でも、ポーリングの関係でサーバAは4時0分10秒に実行し、サーバBは4時0分50秒に実行しようとするかもしれませんので、NOW()で単純に比較すると危険です。大雑把なシステムであれば、30分ぐらいのマージンをNOW()に足したり引いたりすれば適当に計算することもできるでしょうが、境界合わせもプログラムで書けばそれほど難しくはないはずです。

単純なスケジューラであればこれでいいのですが、実行時障害からの復旧や、毎時や毎週などの複雑な設定を可能にする場合には少しカラムを変えたほうがいいでしょう。

改良スケジューラ

-- 作成
CREATE TABLE cron_table (
  id INTEGER NOT NULL,
  hour INTEGER NOT NULL,
  command VARCHAR(255) NOT NULL,
  exec_by VARCHAR(32), -- マーキング用
  next_exec TIMESTAMP NOT NULL -- 次回実行時間
);
-- タスク登録
INSERT lock_table VALUES(1, 4, 'echo 4時になったよ', NULL, '1970/01/01', 0, 3);
-- タスク取得(実行権限取得を兼ねる)
UPDATE cron_table SET exec_by = 'HOSTNAME', next_exec = NOW() + INTERVAL 5 MINUTES WHERE next_exec <= NOW();
SELECT * FROM cron_table WHERE exec_by = 'HOSTNAME';
-- タスク解放(2015年12月13日4時の実行)
UPDATE cron_table SET exec_by = NULL, next_exec = '2015/12/14 04:00';

last_execをnext_execに変えることで、期限が来たすべてのタスクをシンプルにSQLで取得できるようになります。たとえば4時の間にサーバが止まっていたとしても、5時に復旧した時点で実行されるようになります。
なにより例では毎日n時の実行タスクに限定していますが、毎時や毎週などの複雑な設定でも、next_execを見るだけで済みます。last_execの場合は、実行間隔ごとのSELECTか、複雑なSQLが必要になるはずです。next_execであれば、タスク解放時にプログラム側でその実行時間を計算することで、複雑さを封じ込めることができます。

next_execをスケジュール実行時刻だけでなく、障害時のリトライ時刻としても利用していることには異論があるかもしれません。SQL上、それが最も楽な構成になるはずですが、リトライ時刻はカラムとしてみても良いとは思います。

設計ガイド

スケジューラは細かく見るともっといろいろ考えることがあります。最初のスケジューラでは、最初の実行は初期値を入れた次の4時です。改良スケジューラでは次の4時を待たず、ただちに実行されてしまいます（次の4時にも実行されます）。
スケジュールされた時間に実行不可能状態にあった場合、回復後にその分の実行を何回するべきかどうかや、実行中に落ちたかフリーズした場合に何回再実行を試みるべきか（たとえば、そのタスクが誤ったコマンドを登録されているかもしれません）によって設計は変化するはずですので、要件をきちんと考えるようにしましょう。これは、MySQLを使わない一般的なスケジューラでも同じです。

キュー

サーバ間キューことジョブキューです。プロセス間をまたいだ時点でジョブキューと言ってしまってよさそうですね。本当に簡易なジョブキューについては説明の必要はないと思います。適当に放り込んだ中からLIMIT付きのUPDATEでレコード単位でロックしてしまえば実現可能です。

-- 作成
CREATE TABLE queue_table (
  id INTEGER NOT NULL,
  job VARCHAR(255) NOT NULL, -- ジョブ内容
  lock BOOLEAN NOT NULL, -- ジョブのロック
  insert_time TIMESTAMP NOT NULL
);
-- 登録
INSERT queue_table VALUES(1, 'do something', FALSE, NOW());
-- 選択
UPDATE queue_table SET lock = TRUE WHERE lock = FALSE ORDER BY insert_time LIMIT 10;
-- 解放
DELETE FROM queue_table WHERE id = ?;

入れた順で処理するようにinsert_timeでORDER BYしています。
これにロックの制限時間対応やマーキング対応をする場合は今までと同様ですね。解放時にレコード消していますが、これは削除フラグでも状態変更でもよいと思います。
しかし、このやり方は意外に書き込みコストが高くなることがあります。本当に簡易でキューサイズが常に十分小さいなら、MEMORYストレージエンジンの使用を検討するとよいでしょう。その場合は、解放時には確実にDELETEすることを忘れずに。

MEMORYストレージエンジンが使えず、DB負荷が高い場合の軽減策を示しておきます。
まず、キューに対する書き込み操作は、登録、ロック、解放の3種類で、それぞれ同じぐらいの負荷だと考えることができます。データのライフサイクルで考えると本質的に不要なのはロックです。したがって、レコード単位のロックを止めて、ブロック単位でロックできれば、レコードのUPDATEが減って負荷は減るはずです。

ブロック単位のロック

そういうわけでブロック単位のロックを考えます。最も楽なのはキューテーブル単位でロックすることです。しかし、キューテーブル単位でロックしてしまうと、ジョブ完了までロックが解放できないので、1つのワーカーしか動くことができなくなります。せっかくのジョブキューなので非常時の冗長系だけでなくマルチワーカーで負荷分散と行きたいところですね。

そこで並列処理ができるようにキューを分割しましょう。それぞれのレコードに分割IDを入れるようにします。分割はハッシュ値やレコードのIDの剰余でもよいでしょうし、乱数でもよいでしょう。MySQLにはRAND()があるため、状況によらず楽なのは乱数だと思いますが、後のトレース性は落ちるかも知れません。分割値はインデックスを利かせたいこともあるので最初からテーブルに入れておきます。とりあえず4分割してみます。

-- 作成
CREATE TABLE queue_table (
  id INTEGER NOT NULL,
  job VARCHAR(255) NOT NULL, -- ジョブ内容
  divided INTEGER NOT NULL, -- 分割値
  insert_time TIMESTAMP NOT NULL
);
CREATE TABLE q_lock_table (
  divided INTEGER NOT NULL, -- 分割値
  lock_by VARCHAR(32), -- プロセスユニーク名
  expire_at TIMESTAMP NOT NULL -- ロック解放時間
);
INSERT IGNORE q_lock_table VALUES (0, null, '1970/01/01'), (1, null, '1970/01/01'), (2, null, '1970/01/01'), (3, null, '1970/01/01');
-- 登録
INSERT queue_table VALUES(1, 'do something', FLOOR(RAND() * 4), NOW());
-- 選択
UPDATE q_lock_table SET lock_by = 'HOSTNAME', expire_at = NOW() + INTERVAL 5 MINUTES WHERE expire_at < NOW() ORDER BY expire_at LIMIT 1;
SELECT * FROM queue_table WHERE divided = (SELECT divided FROM q_lock_table WHERE lock_by = 'HOSTNAME') ORDER BY insert_time LIMIT 100;
-- 解放(ジョブ)
DELETE FROM queue_table WHERE id = ?;
-- 解放(ロック)
UPDATE q_lock_table SET expire_at = NOW() - INTERVAL 1 SECONDS, lock_by = NULL WHERE lock_by = 'HOSTNAME';

ワーカーはまず自分が使う分割値をロックすることで取得します。この例ではexpire_atを流用してLRUで取得していますが、ランダム、または何らかの偏りの少ない順序で決めてもよいと思います。ORDER BYで優先順位を指定することで、ウェイトフリーっぽく、ロックが取得できなかったということがなくなる（ワーカーが分割数以下の場合）ので、ORDER BYで指定できる属性が良いでしょう。
ワーカーは一定個数を処理し終えたらロックを開放して、別の分割値の取得を行うようにします。このときの処理個数分はキューの処理順序の不公平性につながってくるので、要件に応じて多すぎない値を設定します。

分割数はワーカーの数に応じて設計します。この数だけはDB構造から分離したマジックナンバーになりがちです。不格好ですがSELECT COUNT(*) FROM q_lock_tableで求めることでDBだけで完結させることもできます。

さいごに

MySQLを使ってロック・スケジューラ・キューを実現できることを紹介しました。必ずしも正道ではないと思いますが、実際にテーブル設計を行っていると、気が付くとこのような機能を内包していたりすることもあるので、それほど違和感はないと思います。

もちろんid、insert_time、update_timeの付与や必要に応じた正規化は行ってください。また、時間情報をサーバ間でやりとりするときにはタイムゾーンと時間設定の誤差、ネットワークやプロセスタイムスライスによる処理の遅延に配慮するように気を付けてください。

なお、手法は知らないのですが、RubyだとRDBを使ったスケジューラとキューについてはこのような便利そうなOSSがあることを投稿前に知ったので紹介しておきます。
* PerfectSched
* PerfectQueue