Sidekiqで重複ジョブを防ぐ3つの方法 ― 自作実装からsidekiq-unique-jobsまで

Rubyでコードを書いているなら、バックグラウンド処理にはSidekiqを使っている方が多いのではないでしょうか。ActiveJobなど他の仕組みから移行してきた方にも役立つヒントが含まれているので、ぜひ最後まで読んでみてください。

バックグラウンドジョブの活用シーンはさまざまです。数値計算を行うケースもあれば、ユーザーへのウェルカムメール送信、データ同期のスケジューリングなどもあります。どのような用途であれ、いずれ「重複するジョブを避けたい」という要件にぶつかることがあります。ここでいう重複ジョブとは、まったく同じ処理を行う2つ以上のジョブのことです。もう少し詳しく見ていきましょう。

なぜジョブの重複を排除するのか？

次のようなジョブを想像してみてください。

class BookSalesWorker
  include Sidekiq::Worker
 
  def perform(book_id)
    crunch_some_numbers(book_id)
 
    upload_to_s3
  end
 
  ...
end

BookSalesWorkerは常に同じ処理を行います。book_idをもとにDBから書籍を検索し、最新の販売データを取得して数値を計算した後、ストレージサービスへアップロードします。そして、サイト上で書籍が売れるたびに、このジョブがエンキューされるとします。

さて、もし一度に100件の売上が発生したらどうなるでしょう？同じ処理を行うジョブが100個並ぶことになります。「S3への書き込みは気にならないし、キューも混雑していないから問題ない」と思うかもしれません。しかし、「それでもスケールするのか？」という疑問が残ります。

答えは明確に「No」です。売上や書籍の種類が増えていくと、キューには不要な作業がどんどん積み上がります。1冊あたり100個の同一ジョブが存在し、10冊が同時に売れていれば、実際には各書籍につき1個（合計10個）で足りるはずのところ、キューには1000個のジョブが溜まることになります。

では、重複ジョブがキューに溜まるのを防ぐための選択肢をいくつか紹介していきます。

1. DIY（自作）で対応する

外部依存や複雑なロジックを避けたい場合は、自前のソリューションをコードベースに追加する方法があります。本記事の例を実際に試せるサンプルリポジトリを用意しているので、各アプローチごとにリンクを載せておきます。

1-1. フラグ1個によるアプローチ

ジョブをエンキューするかどうかを判断するフラグを1つ追加する方法です。たとえば、Bookテーブルにsales_enqueued_atカラムを追加して管理するイメージです。

module BookSalesService
  def schedule_with_one_flag(book)
    # 最後にジョブがエンキューされてから10分以上経過しているかチェック
    if book.sales_enqueued_at < 10.minutes.ago
      book.update(sales_enqueued_at: Time.current)
 
      BookSalesWorker.perform_async(book.id)
    end
  end
end

この場合、直近のジョブがエンキューされてから10分間は、新しいジョブはエンキューされません。10分が経過すると、sales_enqueued_atを更新して新しいジョブを投入します。

別のやり方として、crunching_salesのようなboolean型のフラグを使う方法もあります。最初のジョブをエンキューする前にtrueに設定し、ジョブ完了時にfalseへ戻します。crunching_salesがfalseになるまで、後続のジョブはすべて拒否されます。

このアプローチは、作成したサンプルリポジトリで実際に試すことができます。

1-2. フラグ2個によるアプローチ

「10分間ジョブの投入をロックする」という挙動に抵抗があるものの、フラグを追加すること自体は許容できる、という方に向いた方法があります。

既存のsales_enqueued_atに加えて、sales_calculated_atというフラグをもう1つ追加します。コードは次のようになります。

module BookSalesService
  def schedule_with_two_flags(book)
    # 現在集計中かどうかをチェック
    if book.sales_enqueued_at <= book.sales_calculated_at
      book.update(sales_enqueued_at: Time.current)
 
      BookSalesWorker.perform_async(book.id)
    end
  end
end
 
class BookSalesWorker
  include Sidekiq::Worker
 
  def perform(book_id)
    crunch_some_numbers(book_id)
 
    upload_to_s3
 
    # 追加した処理
    book.update(sales_calculated_at: Time.current)
  end
 
  ...
end

動作確認の手順はサンプルリポジトリを参照してください。

この方法では、ジョブがエンキューされてから完了するまでの期間を制御できます。この間は新しいジョブを投入できません。ジョブ実行中はsales_enqueued_atの方がsales_calculated_atより大きくなり、ジョブ完了後はsales_calculated_atがより新しい値（大きい値）になるため、新しいジョブが投入できるようになります。

2つのフラグを使うメリットとして、UIに「最終更新日時」を表示できる点も挙げられます。ユーザーはデータの鮮度を把握できるので、一石二鳥です。

フラグ方式のまとめ

必要に迫られてこうしたソリューションを作りたくなる気持ちはわかりますが、筆者としては少し不格好でオーバーヘッドも伴うように感じます。ユースケースがシンプルであれば使ってもよいですが、複雑さが出てきたり物足りなくなったりしたら、他の選択肢を検討することをおすすめします。

フラグ方式の大きなデメリットは、10分間に投入しようとしたジョブがすべて失われることです。一方で大きなメリットは、依存を導入しなくて済む点と、キュー内のジョブ数を素早く減らせる点です。

1-3. キューを走査する方法

もうひとつのアプローチとして、同じジョブが投入されるのを防ぐ独自のロック機構を作る方法があります。対象のSidekiqキューを調べて、そのジョブ（ワーカー）がすでに存在するかを確認するイメージです。コードは次のようになります。

module BookSalesService
  def schedule_unique_across_queue(book)
    queue = Sidekiq::Queue.new('default')
 
    queue.each do |job|
      return if job.klass == BookSalesWorker.to_s &&
        job.args == [book.id]
    end
 
    BookSalesWorker.perform_async(book.id)
  end
end
 
class BookSalesWorker
  include Sidekiq::Worker
 
  def perform(book_id)
    crunch_some_numbers(book_id)
 
    upload_to_s3
  end
 
  ...
end

上記の例では、'default'キューにBookSalesWorkerクラス名を持つジョブが存在するかを確認しています。さらに引数が書籍IDと一致するかもチェックし、同じBook IDを持つBookSalesWorkerジョブがキュー内にあれば、そこで処理を中断して新しいジョブは投入しません。

ただし注意が必要なのは、ジョブを高速に連続投入すると、キューがまだ空の状態で複数のジョブが登録されてしまう可能性がある点です。実際、筆者がローカルで次のようにテストした際に同様の現象が起きました。

100.times { BookSalesService.schedule_unique_across_queue(book) }

こちらもサンプルリポジトリで試せます。

このアプローチの良いところは、必要であればすべてのキューを横断して既存ジョブを検索できる点です。一方のデメリットは、キューが空の状態で大量に投入すると依然として重複が起こりうること、さらにジョブ投入前にキュー内の全ジョブを走査するため、キューのサイズによってはコストが高くなりうる点です。

2. Sidekiq Enterpriseへアップグレードする

個人または組織に予算の余裕があるなら、SidekiqのEnterprise版へのアップグレードという選択肢もあります。月額179ドルから利用でき、重複ジョブを回避する便利な機能が備わっています。残念ながら筆者はSidekiq Enterpriseを持っていませんが、公式ドキュメントは十分に充実しているはずです。次のようなコードだけで、簡単にユニーク（重複なし）なジョブを実現できます。

class BookSalesWorker
  include Sidekiq::Worker
  sidekiq_options unique_for: 10.minutes
 
  def perform(book_id)
    crunch_some_numbers(book_id)
 
    upload_to_s3
  end
 
  ...
end

これだけです。先ほどの「フラグ1個によるアプローチ」セクションで説明したのとほぼ同じ挙動になります。ジョブは10分間ユニークであり、その期間内に同じ引数を持つジョブはスケジュールできません。

ワンライナーで実現できてかなり便利ですね。Sidekiq Enterpriseをお使いで、この機能をご存じなかった方は、この記事がお役に立てたなら嬉しい限りです。とはいえ大多数の方は利用していないと思うので、次のソリューションに進みましょう。

3. sidekiq-unique-jobsに頼る

はい、gemの話が出ます。しかもLuaファイルを含んでいるため、敬遠する方もいるかもしれません。しかし我慢して聞いてください。これは非常にお得なgemなのです。sidekiq-unique-jobs gemには、多数のロック関連およびその他の設定オプションが用意されています。おそらく必要以上にあるでしょう。

すぐに始めるには、Gemfileにsidekiq-unique-jobsを追加してbundleを実行し、ワーカーを次のように設定します。

class UniqueBookSalesWorker
  include Sidekiq::Worker
 
  sidekiq_options lock: :until_executed,
                  on_conflict: :reject
 
  def perform(book_id)
    book = Book.find(book_id)
 
    logger.info "I am a Sidekiq Book Sales worker - I started"
    sleep 2
    logger.info "I am a Sidekiq Book Sales worker - I finished"
 
    book.update(sales_calculated_at: Time.current)
    book.update(crunching_sales: false)
  end
end

選択肢はたくさんありますが、今回はシンプルにこの設定を採用しました。

sidekiq_options lock: :until_executed, on_conflict: :reject

lock: :until_executedにより、最初のUniqueBookSalesWorkerジョブは実行されるまでロックされます。さらにon_conflict: :rejectを指定することで、実行されようとした他のジョブはすべて拒否され、deadキューへ送られます。これにより、前述のDIYの例とほぼ同等のことが実現できます。

DIYの例からの小さな改善点として、「何が起きたか」の一種のログを確認できる点があります。実際の様子を見てみましょう。

5.times { UniqueBookSalesWorker.perform_async(Book.last.id) }

完全に実行されるのは1つのジョブのみで、残りの4つはdeadキューへ送られます。そこから再試行することも可能です。重複ジョブが単純に無視されたこれまでの例とは、ここが異なる点です。

ロックやコンフリクト解決については多くのオプションから選べます。自分のユースケースに合わせて、gemのドキュメントを参照することをおすすめします。

嬉しいポイント：可視化機能

このgemの優れた点は、ロックの状態やキュー内で何が起きたかの履歴を閲覧できることです。必要なのは、config/routes.rbに以下の行を追加するだけです。

# config/routes.rb
require 'sidekiq_unique_jobs/web'

Rails.application.routes.draw do
  mount Sidekiq::Web, at: '/sidekiq'
end

これで通常のSidekiq管理画面に加え、「Locks（ロック）」と「Changelogs（変更履歴）」の2ページが追加されます。とても便利な機能です。

これらすべては、gemがインストール済みですぐに動かせるサンプルプロジェクトで試すことができます。

なぜLuaなのか？

まず断っておくと、筆者はこのgemの作者ではないので、あくまで推測です。初めてこのgemを見たとき、「Rubyのgemの中でなぜLuaを使うのか？」と疑問に思いました。一見奇妙に映りますが、実はRedisはLuaスクリプトの実行をサポートしています。おそらく作者はそれを踏まえ、軽快なロジックをLuaで実装したかったのでしょう。

gemリポジトリ内のLuaファイルを見ると、それほど複雑ではありません。すべてのLuaスクリプトは、後からRubyコード側のSidekiqUniqueJobs::Script::Callerから呼び出されています。ソースコードを読んで仕組みを理解するのは非常に勉強になるので、ぜひ覗いてみてください。

代替gem

ActiveJobを多用している方は、active-job-uniqueness gemを試してみるとよいでしょう。発想は似ていますが、独自のLuaスクリプトではなくRedlockを使ってRedis内のアイテムをロックします。

このgemでユニークなジョブを実現するには、次のようなジョブをイメージしてください。

class BookSalesJob < ActiveJob::Base
  unique :until_executed
 
  def perform
    ...
  end
end

記法はsidekiq-unique-jobs gemよりも簡潔ですが、よく似ています。ActiveJobに強く依存している環境なら、これで解決できるかもしれません。

まとめ

アプリ内で重複ジョブに対処する方法について、何かしらの知見を得られたなら幸いです。筆者自身、さまざまなソリューションを調査・検証する過程を楽しめました。求めていたものが見つからなかったとしても、紹介した例がオリジナルのソリューションを生み出すきっかけになれば嬉しいです。

すべてのコードスニペットはサンプルプロジェクトにまとめてあります。

それではまた次回、お会いしましょう。

P.S. Ruby Magicの記事を公開と同時にお読みになりたい方は、Ruby Magicニュースレターを購読すれば、記事を見逃すことはありません！