Batch background jobs with Sidekiq

Thứ Sáu, 29/01/2021

Tram Ho

When we want to create multiple background jobs that run in parallel and want to know when all of those jobs are complete, what do we do?

Sidekiq Pro allows me to run a set of background jobs in parallel and then receive a callback right after all jobs have been run. However, Sidekiq Pro is Sidekiq’s paid feature. Thankfully, sidekiq-batch gem has been built based on the Sidekiq Pro API and I can use it for free, and the usage is similar to Sidekiq Pro.

Installation

gem <span class="token string">"sidekiq-batch"</span>

1 2	gem <span class="token string">"sidekiq-batch"</span>

$ bundle install

1 2	$ bundle install

Basic Usage

batch <span class="token operator">=</span> <span class="token constant">Sidekiq</span> <span class="token punctuation">:</span> <span class="token punctuation">:</span> <span class="token constant">Batch</span> <span class="token punctuation">.</span> <span class="token keyword">new</span>
batch <span class="token punctuation">.</span> description <span class="token operator">=</span> <span class="token string">"Creating Complex Jobs..."</span>
batch <span class="token punctuation">.</span> on <span class="token punctuation">(</span> <span class="token symbol">:success</span> <span class="token punctuation">,</span> <span class="token constant">MyCallback</span> <span class="token punctuation">,</span> option <span class="token punctuation">)</span> <span class="token comment">#option là dạng Hash</span>
batch <span class="token punctuation">.</span> on <span class="token punctuation">(</span> <span class="token symbol">:complete</span> <span class="token punctuation">,</span> <span class="token constant">MyCallback</span> <span class="token punctuation">,</span> option <span class="token punctuation">)</span> <span class="token comment">#option là dạng Hash</span>
batch <span class="token punctuation">.</span> jobs <span class="token keyword">do</span>
 <span class="token constant">ComplexWorker1</span> <span class="token punctuation">.</span> perform_async
 <span class="token constant">ComplexWorker2</span> <span class="token punctuation">.</span> perform_async
 <span class="token constant">ComplexWorker3</span> <span class="token punctuation">.</span> perform_async
 <span class="token punctuation">.</span> <span class="token punctuation">.</span> <span class="token punctuation">.</span>
<span class="token keyword">end</span>

batch = Sidekiq : : Batch . new

batch . description = "Creating Complex Jobs..."

batch . on ( :success , MyCallback , option ) #option là dạng Hash

batch . on ( :complete , MyCallback , option ) #option là dạng Hash

batch . jobs do

ComplexWorker1 . perform_async

ComplexWorker2 . perform_async

ComplexWorker3 . perform_async

. . .

end

When the set of all jobs is completed, MyCallBack is executed.

<span class="token keyword">class</span> <span class="token class-name">MyCallBack</span>
  <span class="token keyword">def</span> <span class="token method-definition"><span class="token function">on_complete</span></span> <span class="token punctuation">(</span> status <span class="token punctuation">,</span> options <span class="token punctuation">)</span>
    puts <span class="token string">"Batch has failures"</span> <span class="token keyword">if</span> status <span class="token punctuation">.</span> failures <span class="token operator">!=</span> <span class="token number">0</span>
  <span class="token keyword">end</span>
  <span class="token keyword">def</span> <span class="token method-definition"><span class="token function">on_success</span></span> <span class="token punctuation">(</span> status <span class="token punctuation">,</span> options <span class="token punctuation">)</span>
    puts <span class="token string">"Batch succeed"</span>
  <span class="token keyword">end</span>
<span class="token keyword">end</span>

class MyCallBack

def on_complete ( status , options )

puts "Batch has failures" if status . failures != 0

end

def on_success ( status , options )

puts "Batch succeed"

end

complete – when all jobs in the batch have been run at least once, successful or not.
success – when all jobs in the batch have been run successfully.
death – when an error occurs in the batch

In the above callbacks, params status will have the following options:

status <span class="token operator">=</span> <span class="token constant">Sidekiq</span> <span class="token punctuation">:</span> <span class="token punctuation">:</span> <span class="token constant">Batch</span> <span class="token punctuation">:</span> <span class="token punctuation">:</span> <span class="token constant">Status</span> <span class="token punctuation">.</span> <span class="token keyword">new</span> <span class="token punctuation">(</span> bid <span class="token punctuation">)</span>
status <span class="token punctuation">.</span> total <span class="token comment"># số jobs</span>
status <span class="token punctuation">.</span> failures <span class="token comment"># số jobs đã fail</span>
status <span class="token punctuation">.</span> pending <span class="token comment"># số jobs đang chạy chưa thành công</span>
status <span class="token punctuation">.</span> created_at <span class="token comment"># =&gt; datetime chạy</span>
status <span class="token punctuation">.</span> complete <span class="token operator">?</span> <span class="token comment">#  check nếu batch đã complete chưa</span>
status <span class="token punctuation">.</span> failure_info <span class="token comment"># array của jobs failed</span>
status <span class="token punctuation">.</span> data <span class="token comment"># hash của batch data</span>

status = Sidekiq : : Batch : : Status . new ( bid )

status . total # số jobs

status . failures # số jobs đã fail

status . pending # số jobs đang chạy chưa thành công

status . created_at # => datetime chạy

status . complete ? # check nếu batch đã complete chưa

status . failure_info # array của jobs failed

status . data # hash của batch data

Demo Example

<span class="token comment"># app/services/create_complex_jobs_service.rb</span>

<span class="token keyword">class</span> <span class="token class-name">CreateComplexJobsService</span>
  <span class="token keyword">def</span> <span class="token method-definition"><span class="token function">perform</span></span>
    batch <span class="token operator">=</span> <span class="token constant">Sidekiq</span> <span class="token punctuation">:</span> <span class="token punctuation">:</span> <span class="token constant">Batch</span> <span class="token punctuation">.</span> <span class="token keyword">new</span>
    batch <span class="token punctuation">.</span> description <span class="token operator">=</span> <span class="token string">"Creating cluster"</span>
    batch <span class="token punctuation">.</span> on <span class="token punctuation">(</span> <span class="token symbol">:success</span> <span class="token punctuation">,</span> <span class="token constant">ComplexJobCallbackService</span> <span class="token punctuation">)</span>
    batch <span class="token punctuation">.</span> on <span class="token punctuation">(</span> <span class="token symbol">:complete</span> <span class="token punctuation">,</span> <span class="token constant">ComplexJobCallbackService</span> <span class="token punctuation">)</span>
    batch <span class="token punctuation">.</span> jobs <span class="token keyword">do</span>
      <span class="token number">5.</span> times <span class="token punctuation">{</span> <span class="token operator">|</span> i <span class="token operator">|</span> <span class="token constant">CreateComplexJobWorker</span> <span class="token punctuation">.</span> perform_async <span class="token punctuation">(</span> i <span class="token punctuation">)</span> <span class="token punctuation">}</span>
    <span class="token keyword">end</span>
  <span class="token keyword">end</span>
<span class="token keyword">end</span>

# app/services/create_complex_jobs_service.rb

class CreateComplexJobsService

def perform

batch . description = "Creating cluster"

batch . on ( :success , ComplexJobCallbackService )

batch . on ( :complete , ComplexJobCallbackService )

batch . jobs do

5. times { | i | CreateComplexJobWorker . perform_async ( i ) }

end

<span class="token comment"># app/services/complex_job_callback_service.rb</span>

<span class="token keyword">class</span> <span class="token class-name">ComplexJobCallbackService</span>
  <span class="token keyword">def</span> <span class="token method-definition"><span class="token function">on_success</span></span> <span class="token punctuation">(</span> status <span class="token punctuation">,</span> options <span class="token punctuation">)</span>
    puts <span class="token string">"----"</span>
    puts status <span class="token punctuation">,</span> options
    puts <span class="token string">"Batch success"</span>
  <span class="token keyword">end</span>

  <span class="token keyword">def</span> <span class="token method-definition"><span class="token function">on_complete</span></span> <span class="token punctuation">(</span> status <span class="token punctuation">,</span> options <span class="token punctuation">)</span>
    <span class="token keyword">if</span> status <span class="token punctuation">.</span> failures <span class="token operator">!=</span> <span class="token number">0</span>
      puts <span class="token string">"Batch has failures"</span>
      puts status <span class="token punctuation">.</span> failure_info
    <span class="token keyword">end</span>
    puts <span class="token string">"----"</span>
    puts status <span class="token punctuation">,</span> options
    puts <span class="token string">"Batch complete"</span>
  <span class="token keyword">end</span>
<span class="token keyword">end</span>

# app/services/complex_job_callback_service.rb

class ComplexJobCallbackService

puts "----"

puts status , options

puts "Batch success"

end

if status . failures != 0

puts "Batch has failures"

puts status . failure_info

end

puts "----"

puts status , options

puts "Batch complete"

end

<span class="token comment"># app/workers/create_complex_job_worker.rb</span>

<span class="token keyword">class</span> <span class="token class-name">CreateComplexJobWorker</span>
  <span class="token keyword">include</span> <span class="token constant">Sidekiq</span> <span class="token punctuation">:</span> <span class="token punctuation">:</span> <span class="token constant">Worker</span>
  <span class="token keyword">def</span> <span class="token method-definition"><span class="token function">perform</span></span> <span class="token punctuation">(</span> id <span class="token punctuation">)</span>
    puts <span class="token string">"Creating job <span class="token interpolation"><span class="token delimiter tag">#{</span> id <span class="token delimiter tag">}</span></span> ..."</span>
    sleep <span class="token number">1</span>
  <span class="token keyword">end</span>
<span class="token keyword">end</span>

# app/workers/create_complex_job_worker.rb

class CreateComplexJobWorker

include Sidekiq : : Worker

def perform ( id )

puts "Creating job #{ id } ..."

sleep 1

end

Result:

$ CreateComplexJobsService.new.perform
=&gt; ["08353e8f349b8452800e922c", "55444a8b49e196ebf91d1003", "ab31baab81a49a7773fb056d", "474e88d1442faf7e6159a077", "8300a336f66a6f8c72e85adc"]

$ CreateComplexJobsService.new.perform

=> ["08353e8f349b8452800e922c", "55444a8b49e196ebf91d1003", "ab31baab81a49a7773fb056d", "474e88d1442faf7e6159a077", "8300a336f66a6f8c72e85adc"]

2021-01-22T08:16:59.356Z 4220 TID-ovfhvkj7o CreateComplexJobWorker JID-08353e8f349b8452800e922c BID-tdYhwcrQsEQabA INFO: start
2021-01-22T08:16:59.359Z 4220 TID-ovfhf08hs CreateComplexJobWorker JID-ab31baab81a49a7773fb056d BID-tdYhwcrQsEQabA INFO: start
2021-01-22T08:16:59.363Z 4220 TID-ovfhf08gs CreateComplexJobWorker JID-55444a8b49e196ebf91d1003 BID-tdYhwcrQsEQabA INFO: start
2021-01-22T08:16:59.365Z 4220 TID-ovfhvkizk CreateComplexJobWorker JID-8300a336f66a6f8c72e85adc BID-tdYhwcrQsEQabA INFO: start
2021-01-22T08:16:59.371Z 4220 TID-ovfhf07kg CreateComplexJobWorker JID-474e88d1442faf7e6159a077 BID-tdYhwcrQsEQabA INFO: start
Creating job 0...
2021-01-22T08:17:00.440Z 4220 TID-ovfhvkj7o CreateComplexJobWorker JID-08353e8f349b8452800e922c BID-tdYhwcrQsEQabA INFO: done: 1.084 sec
Creating job 2...
Creating job 3...
Creating job 1...
Creating job 4...
2021-01-22T08:17:01.491Z 4220 TID-ovfhf08hs CreateComplexJobWorker JID-ab31baab81a49a7773fb056d BID-tdYhwcrQsEQabA INFO: done: 2.131 sec
2021-01-22T08:17:01.492Z 4220 TID-ovfhf07kg CreateComplexJobWorker JID-474e88d1442faf7e6159a077 BID-tdYhwcrQsEQabA INFO: done: 2.121 sec
2021-01-22T08:17:01.495Z 4220 TID-ovfhvkizk CreateComplexJobWorker JID-8300a336f66a6f8c72e85adc BID-tdYhwcrQsEQabA INFO: done: 2.129 sec
2021-01-22T08:17:01.497Z 4220 TID-ovfhvkj7o Sidekiq::Batch::Callback::Worker JID-c7f98aa5721f57015a48334b BID-5UNzECYjGM035Q INFO: start
2021-01-22T08:17:01.497Z 4220 TID-ovfhf08gs CreateComplexJobWorker JID-55444a8b49e196ebf91d1003 BID-tdYhwcrQsEQabA INFO: done: 2.134 sec
----
#&lt;Sidekiq::Batch::Status:0x00007fa34b6bdd98&gt;
{}
Batch complete
2021-01-22T08:17:01.501Z 4220 TID-ovfhf07rg Sidekiq::Batch::Callback::Worker JID-c57ca2bcd2fe25269066a8ea BID-GJJ4t5EPp_XGNw INFO: start
2021-01-22T08:17:01.501Z 4220 TID-ovfhvkj7o Sidekiq::Batch::Callback::Worker JID-c7f98aa5721f57015a48334b BID-5UNzECYjGM035Q INFO: done: 0.005 sec
----
#&lt;Sidekiq::Batch::Status:0x00007fa3474ebed8&gt;
{}
Batch success
2021-01-22T08:17:01.504Z 4220 TID-ovfhf07rg Sidekiq::Batch::Callback::Worker JID-c57ca2bcd2fe25269066a8ea BID-GJJ4t5EPp_XGNw INFO: done: 0.003 sec

2021-01-22T08:16:59.356Z 4220 TID-ovfhvkj7o CreateComplexJobWorker JID-08353e8f349b8452800e922c BID-tdYhwcrQsEQabA INFO: start

2021-01-22T08:16:59.359Z 4220 TID-ovfhf08hs CreateComplexJobWorker JID-ab31baab81a49a7773fb056d BID-tdYhwcrQsEQabA INFO: start

2021-01-22T08:16:59.363Z 4220 TID-ovfhf08gs CreateComplexJobWorker JID-55444a8b49e196ebf91d1003 BID-tdYhwcrQsEQabA INFO: start

2021-01-22T08:16:59.365Z 4220 TID-ovfhvkizk CreateComplexJobWorker JID-8300a336f66a6f8c72e85adc BID-tdYhwcrQsEQabA INFO: start

2021-01-22T08:16:59.371Z 4220 TID-ovfhf07kg CreateComplexJobWorker JID-474e88d1442faf7e6159a077 BID-tdYhwcrQsEQabA INFO: start

Creating job 0...

2021-01-22T08:17:00.440Z 4220 TID-ovfhvkj7o CreateComplexJobWorker JID-08353e8f349b8452800e922c BID-tdYhwcrQsEQabA INFO: done: 1.084 sec

Creating job 2...

Creating job 3...

Creating job 1...

Creating job 4...

2021-01-22T08:17:01.491Z 4220 TID-ovfhf08hs CreateComplexJobWorker JID-ab31baab81a49a7773fb056d BID-tdYhwcrQsEQabA INFO: done: 2.131 sec

2021-01-22T08:17:01.492Z 4220 TID-ovfhf07kg CreateComplexJobWorker JID-474e88d1442faf7e6159a077 BID-tdYhwcrQsEQabA INFO: done: 2.121 sec

2021-01-22T08:17:01.495Z 4220 TID-ovfhvkizk CreateComplexJobWorker JID-8300a336f66a6f8c72e85adc BID-tdYhwcrQsEQabA INFO: done: 2.129 sec

2021-01-22T08:17:01.497Z 4220 TID-ovfhvkj7o Sidekiq::Batch::Callback::Worker JID-c7f98aa5721f57015a48334b BID-5UNzECYjGM035Q INFO: start

2021-01-22T08:17:01.497Z 4220 TID-ovfhf08gs CreateComplexJobWorker JID-55444a8b49e196ebf91d1003 BID-tdYhwcrQsEQabA INFO: done: 2.134 sec

----

#<Sidekiq::Batch::Status:0x00007fa34b6bdd98>

{}

Batch complete

2021-01-22T08:17:01.501Z 4220 TID-ovfhf07rg Sidekiq::Batch::Callback::Worker JID-c57ca2bcd2fe25269066a8ea BID-GJJ4t5EPp_XGNw INFO: start

2021-01-22T08:17:01.501Z 4220 TID-ovfhvkj7o Sidekiq::Batch::Callback::Worker JID-c7f98aa5721f57015a48334b BID-5UNzECYjGM035Q INFO: done: 0.005 sec

----

#<Sidekiq::Batch::Status:0x00007fa3474ebed8>

{}

Batch success

2021-01-22T08:17:01.504Z 4220 TID-ovfhf07rg Sidekiq::Batch::Callback::Worker JID-c57ca2bcd2fe25269066a8ea BID-GJJ4t5EPp_XGNw INFO: done: 0.003 sec

As the result above, we see that after the jobs ran in complete parallel, the on_success and on_complete callbacks performed as expected.

References:

https://github.com/breamware/sidekiq-batch

https://github.com/mperham/sidekiq/wiki/Batches

https://gorails.com/episodes/batching-background-jobs-with-sidekiq?autoplay=1

Chia sẻ bài viết ngay

Nguồn bài viết : Viblo

Batch background jobs with Sidekiq

Installation

Basic Usage

Demo Example

TikTok trở thành nền tảng mảng xã hội lớn thứ hai ở Nam Phi

Mất giá nhanh nhất sau 9 tháng ra mắt, iPhone 14 Pro Max tiếp tục phá đáy tại Việt Nam

Hướng dẫn sử dụng ngôn ngữ R cho người mới bắt đầu

10 plugin cần thiết của SublimeText dành cho các lập trình viên JavaScript