sidekiq の導入

Rails の非同期処理を扱うとき、これまで delayd_job を使ってきました。
実行内容の保存先として SQL サーバー を使う delayd_job。
対して、実行内容の保存先に Redis を使う sidekiq といった様子のようです。

参考

sidekiq の導入

インストール

Gemfile
1
2
gem 'redis-namespace'
gem 'sidekiq'

アクティブジョブとの連携

active_job の queue_adapter に sidekiq を登録します。
この辺りは delayed_job と同じ感じ。

config/application.rb
1
2
3
4
5
6
7
8
module App
  class Application < Rails::Application
    config.load_defaults 6.1

    # active_job の queue_adapter に :sidekiq を登録
    config.active_job.queue_adapter = :sidekiq
  end
end

接続先 redis 設定

config/initializers/sidekiq.rbに接続先の redis サーバーとクライアントの接続先を記述します。
(redis という名前で redis-server の起動しているホストを docker で用意してあります。)

config/initializers/sidekiq.rb
1
2
3
4
5
6
7
Sidekiq.configure_server do |config|
  config.redis = { url: 'redis://redis:6379', namespace: 'sidekiq_job' }
end

Sidekiq.configure_client do |config|
  config.redis = { url: 'redis://redis:6379', namespace: 'sidekiq_job' }
end

sidekiq 設定

config/sidekiq.ymlを準備。

config/sidekiq.yml
1
2
3
4
5
6
7
8
:verbose: false
:pidfile: ./tmp/pids/sidekiq.pid
:logfile: ./log/sidekiq.log
#concurrencyは同時実行数
#sidekiqの親プロセスも含むそうなので、基本2以上に設定するのでは？
:concurrency: 25
:queues:
  - default

ジョブの作成

次のコマンドでジョブを作成。

1
2
3
4
5
rails generate job Hoge
Running via Spring preloader in process 8465
      invoke  test_unit
      create    test/jobs/hoge_job_test.rb
      create  app/jobs/hoge_job.rb
app/jobs/hoge_job.rb
1
2
3
4
5
6
7
class HogeJob < ApplicationJob
  queue_as :default

  def perform(*args)
    #　遅延処理させたいものを記述する
  end
end

sidekiq の起動

以下コマンドで、sidekiq を起動。

1
bundle exec sidekiq -q default

動作確認

動作確認したいので、app/jobs/hoge_job.rbに確認用の記述を足します。

app/jobs/hoge_job.rb[確認用puts追加]
1
2
3
4
5
6
7
class HogeJob < ApplicationJob
  queue_as :default

  def perform(*args)
    puts "start sidekiq job"
  end
end

Rails コンソールで HogeJob.perform_later を実行すると以下のように sidekiq 側のコンソールに表示されます。

1
2
3
2021-04-25T09:39:03.718Z pid=595 tid=bpb class=HogeJob jid=37c61e2e607ce9e6879f6415 INFO: start
start sidekiq job
2021-04-25T09:39:03.762Z pid=595 tid=bpb class=HogeJob jid=37c61e2e607ce9e6879f6415 elapsed=0.044 INFO: done

遅延ジョブを呼び出すことができました。

遅延処理で ActiveRecord を書き換える

とりあえず実装

今度は sidekiq を使って遅延処理を行い、ActiveRecord のモデルの処理完了時刻を書き込んでみます。

こういうモデルがあるとします。

app/models/item.rb
1
2
3
4
5
6
class Item < ApplicationRecord
  def task_end
    self.task_ended_at = Time.current
    self.save
  end
end

こちらを遅延処理で呼び出してみます。

app/jobs/hoge_job.rb
1
2
3
4
5
6
7
8
9
10
class HogeJob < ApplicationJob
  queue_as :default

  def perform(item)

    # 遅延実行したい何か

    item.task_end
  end
end

次のように実行するとitem.task_endが実行されて、時刻の書き込みがされます。

1
2
i = Item.create
HogeJob.perform_later(i)

ですが、mperham/sidekiq - Getting Startedには、以下の記述があります。

Your perform method arguments must be simple, basic types like String, integer, boolean that are supported by JSON. Complex Ruby objects will not work.

perform メソッドの引数は、string・integer・bool のような JSON によってサポートされている単純なタイプでなければなりません。複雑な Ruby オブジェクトは機能しないそうです。

ActiveRecord のオブジェクトは、単純なタイプとは言えないので、直してみます。

単純なオブジェクトを引数に実行させる

ID だけを渡して実行できるようにします。

app/jobs/hoge_job.rb[修正１]
1
2
3
4
5
6
7
8
9
10
class HogeJob < ApplicationJob
  queue_as :default

  def perform(target_id)

    # 遅延実行したい何か

    Item.find(target_id).task_end
  end
end

呼び出しは次のように変わります。

1
2
i = Item.create
HogeJob.perform_later(i.id)

単純な integer を渡すだけで遅延ジョブで ActiveRecord を書き換えできました。

一般化してみる

sidekiq のドキュメントに基づいて修正をしましたが、このままだと書き換えするモデルが変わる・増える度に、実装の修正・追加が必要になります。

実行内容を示す、クラス名、メソッド、対象の ID をすべてパラメーターで渡してみます。

app/jobs/hoge_job.rb[修正１]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
class HogeJob < ApplicationJob
  queue_as :default

  def perform(target_class, target_methods, target_id)

    # 遅延実行したい何か

    # target_class が 名前空間にあればクラスオブジェクトを呼び出せる
    klass = target_class.constantize

    # クラスオブジェクトが、メソッドを持っているか調べる
    klass.respond_to?(target_methods)

    # 指定のクラスでクラスインスタンスを作成
    ins = klass.find(target_id)

    # メソッドを呼び出し
    ins.method(target_methods).call
  end
end

呼び出しは次のようになります。

1
2
i = Item.create
HogeJob.perform_later("Item", :task_end, i.id)

今回はシンプルにするため行っていませんが、クラスとして呼び出せるかのチェックの厳密化や、パラメータを拡張することもできます。
コールバック的に処理を呼び出すクラスとして切り出してしまうほうが、より利便性が高まると考えます。

sidekiq を使ってみました。
遅延ジョブの処理は今まで delayd_job を使ってきましたが、1 つ所にこだわらないようしたいところです。
.delayで呼び出せば良かった delayd_job と比べれば、何を遅延ジョブにしたいのか設計するというステップがしっかり必要そうです。

ではでは。