等待 SuckerPunch::Job 任务完成答案

【问题标题】：Wait for SuckerPunch::Job task completion等待 SuckerPunch::Job 任务完成
【发布时间】：2021-11-10 07:25:11
【问题描述】：

我尝试使用Sucker Punch gem 来并行处理任务。但我没有找到如何等待其终止的文档。

require 'sucker_punch'

class SuckerJob
  include SuckerPunch::Job
  workers 4

  def perform(event)
    sleep(rand(5))
    puts "[#{Thread.current.object_id}] End processing event   #{event}."
  end
end

10.times { |i| SuckerJob.perform_async(i) }
puts "Shutting down ..."
SuckerPunch::Queue.shutdown_all
puts "Shutdown finished, status: #{SuckerPunch::Queue.stats[SuckerJob.name]}"
# [Ugly] call internal method
SuckerPunch::Queue::QUEUES.fetch_or_store(SuckerJob.name).wait_for_termination(10)
puts "Wait finished, status: #{SuckerPunch::Queue.stats[SuckerJob.name]}"

似乎SuckerPunch::Queue.shutdown_all() 在所有任务完成之前返回。

Shutting down ...
[17487240] End processing event   1.
[17488760] End processing event   0.
[17487240] End processing event   4.
[17488760] End processing event   5.
[17486120] End processing event   2.
[17484940] End processing event   3.
[17487240] End processing event   6.
Shutdown finished, status: {"workers"=>{"total"=>3, "busy"=>3, "idle"=>0}, "jobs"=>{"processed"=>7, "failed"=>0, "enqueued"=>0}}
[17484940] End processing event   9.
[17488760] End processing event   7.
[17486120] End processing event   8.
Wait finished, status: {"workers"=>{"total"=>0, "busy"=>0, "idle"=>0}, "jobs"=>{"processed"=>10, "failed"=>0, "enqueued"=>0}}

如何才能等到所有任务都完成？

【问题讨论】：

你可以试试SuckerPunch::Counter::Processed.new(@queue).value
或者检查你可以做的统计all_stats = SuckerPunch::Queue.stats然后stats = all_stats[MyJob.to_s]然后你现在得到的统计你可以看到stats["jobs"]["processed"]现在有total，processed，busy....

标签： ruby concurrency sucker-punch

【解决方案1】：

您可以查看状态

或检查你可以做的统计数据

all_stats = SuckerPunch::Queue.stats

然后

stats = all_stats[SuckerJob.to_s]

然后你现在可以看到统计数据

stats["jobs"]["processed"]

你可以像下面这样检查

  stats["jobs"]["processed"] > 0
  stats["jobs"]["failed"] == 0
   stats["jobs"]["enqueued"] == 0

【讨论】：

【解决方案2】：

我用这个：

def wait_for_jobs(job_name:, count:, max_seconds: 100)
  Rails.logger.info "Waiting up to #{max_seconds} seconds for #{count} jobs to run"
  wait_time = 0
  while wait_time < max_seconds
    stats = SuckerPunch::Queue.stats[job_name]
    processed = stats['jobs']['processed']
    break unless processed < count

    sleep(1)
    wait_time += 1
  end
  raise StandardError, "Timeout while waiting for #{count} jobs of #{job_name} to have run!" unless wait_time < max_seconds

  Rails.logger.info "#{count} jobs took #{wait_time} seconds to run"
end

【讨论】：