Redis のキューに定期的に複数プロセスから重複なしで値を追加

4143 ワード
Redis Ruby Ruby テキストリンク

resque のキューに定期的にタイムスタンプを追加して、cron 的に使いたかったのだが、キューに追加する処理を 1 つのプロセスでやるのは可用性が低いので、(複数のサーバに分散した) 複数のプロセスから重複なしで値を追加したかった。

下記の方法でうまくいきそう。

  1. 最後に値を追加したときのタイムスタンプを保持しておき、追加前にチェックする
  2. 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 が使える。