resque のキューに定期的にタイムスタンプを追加して、cron 的に使いたかったのだが、キューに追加する処理を 1 つのプロセスでやるのは可用性が低いので、(複数のサーバに分散した) 複数のプロセスから重複なしで値を追加したかった。
下記の方法でうまくいきそう。
- 最後に値を追加したときのタイムスタンプを保持しておき、追加前にチェックする
- 1 だけだとほぼ同時に実行される場合に重複する可能性があるので Redis のトランザクション機構を利用する
下記は 1 秒ごとにタイムスタンプをキューに追加する例。
# cron.rb
require "redis"
require "json"
QUEUE_NAME = "per_sec" # キューの名前
TIMESTAMP_NAME = "per_sec_timestamp" # 最後にキューに値を追加したときのタイムスタンプ
redis = Redis.new
loop do
t = Time.now.to_i
redis.watch(TIMESTAMP_NAME) # TIMESTAMP_NAME の値が exec までに変わっていれば multi - exec 間のコマンドを失敗させる
last = redis.get(TIMESTAMP_NAME).to_i
if last < t
redis.multi
redis.rpush(QUEUE_NAME, t)
redis.set(TIMESTAMP_NAME, t)
redis.exec ? puts("set #{t}") : puts("transaction fail")
end
sleep 0.1
end
上記のスクリプトを 2 プロセス立ち上げて、while true; do redis-cli lpop per_sec; sleep 0.1s; done でキューの内容を pop しつづけると下記のようになり、重複なしでキューに追加できていることがわかる。
Redis のクラスターモードへの対応
クラスターモードの場合 MULTI - EXEC 間の操作はキーが同じハッシュスロットにある場合に限定される (参考)。
これに対応するにはタイムスタンプを保存するキーのハッシュに使う文字列をキューのキー名にすればよい。
TIMESTAMP_KEY = "{#{QUEUE_KEY}}:updated_at".freeze
CLUSTER KEYSLOT per_sec と CLUSTER KEYSLOT {per_sec}:updated_at が一致する = 同じハッシュスロットに保存されるので MULTI - EXEC が使える。