PostgreSQL で簡易に MQ

「PostgreSQL で安くて早くてうまい MQ 作ってね」的ミッションがどこからともなく降ってきたので、色々調べて試してみたところ、案外よい方法が見つかったので、まとめてみた。

• [4/24] 複数キューを扱う場合の方法を tableoid を利用するように更新。
• [4/27] フィルタ条件を付ける場合の記法について追記(WHERE 句内の評価順の明示的指定)
• [4/30] 優先順位に ORDER BY は利用できない旨追記

鍵は advisory lock。PostgreSQL 8.2 以降の機能だ。

メッセージキューと言っても、安くて早いわけだからもちろんフルフルの高機能なわけなく、いろいろ制限がある。

• receiver は複数可。メッセージは receiver の一つが受信する。ブロードキャスト/マルチキャストは対象外。
• ack あり。ack せずに receiver が落ちた場合、メッセージは別の receiver が受信可能になる。
• receiver は PULL でメッセージを受信*1
• メッセージ処理に長時間かからない（メッセージ受信から ack 送信まで、DBセッションを握ったままでよい）

つまり、排他性と receiver の１つに障害が発生しても処理が行われるという要件を満たすことを優先している。

メッセージを格納するテーブル

create table mq (
    id serial, 
    message text, 
    primary key(id)
);

必須フィールドは id。serial 。
キューを複数を持ちたい場合は、それぞれ対応するテーブルを作成。
必要なデータ・メタデータがあれば適宜フィールドを追加。

キューにメッセージを送信(send)

insert into mq (text) values ('abc');

そのまま insert でＯＫ。

キューからメッセージを１件取得(receive)

下記 SQL によりポーリング(PULL)。

select * from mq where pg_try_advisory_lock(tableoid::int, id) limit 1; 

PostgreSQL の advisory lock はアプリが独自に利用可能なロック管理機能。

pg_try_advisory_lock(tableoid::int, id) は (tableoid, id) をキーにして advisory lock の取得に成功したら true、失敗したら false を返すので、この１行で

「対象キューテーブルの、ロックされていないメッセージを最大１件取得し、id をキーにロックする」

という動作になる。
受け取り可能なメッセージがなければ、０件。

テーブルに適当なメタデータ追加＆ receive の SQL に ORDER BY や WHERE 句追加でメッセージの優先順位やフィルタも実現できる。宛先指定もなんとかなるかな。

ただし、WHERE 句内に AND で並べた式の評価順は保証されないので、副作用を持つ pg_try_advisory_lock() を含めてそのまま並べただけでは、意図したとおりに動作するかどうかは断言できない。
例えばフィルタ条件より先に pg_try_advisory_lock() を評価されてしまったら、取り出すことができたメッセージ以外にもロックかけまくってる、なんて現象が発生してしまう*2。

そういう場合には CASE 文を使って明示的に評価順を記述する方法がある、と PostgreSQL のマニュアルに書いてあった。
ので、「where case when [フィルタ条件] then pg_try_advisory_lock() else false end」のように記述すればよいだろう。ちょっと BK くさいけど。

select * from mq
  where case when priority < 10 then pg_try_advisory_lock(tableoid::int, id) else false end
  limit 1; 

また、優先順位の処理に ORDER BY が利用できるか、と考えていたのだが、id:sfujiwara さんのご指摘により、ORDER BY 句をつけると並べ替えのために全てのレコードに対して WHERE 条件を評価してしまい、全レコード分のロックを取得してしまうことがわかった。
サブクエリなどを駆使したり、volatile で関数宣言したりして回避できないか色々試してみたがダメ。
というわけで、優先順位などで並べ替えをやりたければ、カーソルで１レコードずつ取って pg_try_advisory_lock() するしかなさそう。残念。

更新ロックをかけたければ for update をつけられるけど、advisory lock によりアプリレイヤーでは排他性は確保できているので、トランザクション不要でいいんじゃあないかと*3。

なお、advisory lock は transaction と無関係。rollback でも解放されない。
が、セッションが切れれば解放される。

メッセージ処理完了(ack)

delete from mq where id=12345
  returning pg_advisory_unlock(tableoid::int, id); 

メッセージを削除しつつ、advisory lock を解放。成功すれば true が返る。
削除したくなければ完了フラグ的なものを用意して、DELETE のかわりに UPDATE を使えばＯＫ。

ack せずにセッションが切れた場合は advisory lock が勝手に解放されて、別の receiver が受信可能に(←ここが嬉しい)。

メッセージのキャンセル(cancel)

メッセージをキャンセルしたいなんて場合もあるかもしれない。

delete from mq where id=12345 and pg_try_advisory_lock(tableoid:int, id)
  returning pg_advisory_unlock(tableoid:int, id); 

キャンセルに成功すれば true、失敗すれば false 、メッセージがすでに処理されてしまっていたら０件が返る。

制限

難点は、advisory lock のキーは 32bit integer×２(または 64bit) なので、利用するアプリ間であらかじめ調整していないと衝突の可能性があること。
とはいえ、advisory lock のキーは database ごとに与えられる(PostgreSQL の lock 機構は、キーに database oid を含む)から、同じデータベースを利用するアプリ間でさえ調整できていればＯＫなので、なんとか現実的な範囲だろう。*4

また、advisory lock はセッション単位なので、 ack するまではセッションを握っておかないといけない。
メッセージを処理している間、コネクション閉じたりプールに返したりしたらダメ。

余談１

「Q4M 使いたいけど、うちポスグレやねんなあ」という人にはちょっと嬉しい？
Q4M だったらメッセージを待っている間ブロッキングしてくれるんだけどね……

余談２

最初、 "UPDATE 〜 SET lock=1 WHERE lock=0 RETURNING *;" でいけんじゃあないかと思っていたら、PostgreSQL の UPDATE 文に LIMIT がなくて……

*1:LISTEN & NOTIFY でタイミングの制御くらいなら別途やってもいいかも

*2: 値の大小比較と pg_try_advisory_lock() なら、比較の方がコストが低い(と推測できる)から先に評価してくれるだろう、とおおむね期待できるんじゃあないかとは思うんだけど

*3:advisory lock と row lock 両方かけて pg_locks 覗いたら、両方はいらんなあという気持ちになれる

*4:名前空間があれば良かったんだけどねー。32bit 整数の片方を enum 的に使う手もあるけど、どっちを識別用に用いるかとかやっぱりアプリ間での調整が必須