ruby-on-rails - 如何为 ActiveRecord 模型对象找到关联的 Resque 作业？

Question

我需要能够找到模型对象的排队和/或工作作业和/或失败作业，例如当模型对象被销毁时我们想要找到所有作业并决定不删除或销毁作业(有条件的)。

在我重新发明轮子之前，是否有推荐的方法来执行此操作？

例子:

如果你想创建一个 before_destroy 回调，当对象被销毁时销毁所有作业(排队和失败的作业)，并且只有在没有工作作业时才销毁

我想为这个示例用例做的一些伪代码:

报表模型

class Report < ActiveRecord::Base
  before_destroy :check_if_working_jobs, :destroy_queued_and_failed_jobs

  def check_if_working_jobs
    # find all working jobs related to this report object
    working_jobs = ProcessReportWorker.find_working_jobs_by_report_id(self.id) 
    return false unless working_jobs.empty?
  end

  def destroy_queued_and_failed_jobs
    # find all jobs related to this report object
    queued_jobs = ProcessReportWorker.find_queued_jobs_by_report_id(self.id) 
    failed_jobs = ProcessReportWorker.find_failed_jobs_by_report_id(self.id) 

    # destroy/remove all jobs found
    (queued_jobs + failed_jobs).each do |job| 
       # destroy the job here ... commands?
    end

  end
end

resque/redis 支持的作业的报告处理 worker 类

class ProcessReportWorker

  # find the jobs by report id which is one of the arguments for the job?
  # envisioned as separate methods so they can be used independently as needed

  def self.find_queued_jobs_by_report_id(id)
     # parse all jobs in all queues to find based on the report id argument?
  end 

  def self.find_working_jobs_by_report_id(id)
     # parse all jobs in working queues to find based on the report id argument? 
  end 

  def self.find_failed_jobs_by_report_id(id)
     # parse all jobs in failed queue to find based on the report id argument? 
  end 
end

这种方法是否符合需要发生的事情？

通过模型对象 id 找到排队或工作的作业然后销毁它上面缺少的部分是什么？

是否已经有方法可以通过我在文档或搜索中遗漏的关联模型对象 ID 来查找和/或销毁？

更新:修改了用法示例，仅使用 working_jobs 作为检查我们是否应该删除的方式，而不是建议我们也尝试删除 working_jobs。 (因为删除工作作业比简单地删除 redis 键条目更复杂)

Answer 1

这里很安静，没有任何回应，所以我设法按照我在问题中指出的路径自己解决了这一天。可能有更好的解决方案或其他方法可用，但到目前为止，这似乎已经完成了工作。如果此处使用的方法有更好的选择，或者是否可以进一步改进，请随时发表评论。

这里的总体方法是您需要搜索所有作业(排队的、工作的、失败的)并仅过滤出相关的 class 和 queue 作业与您在 args 数组的正确索引位置中查找的对象记录 ID 相匹配。例如(在确认 class 和 queue 匹配后)如果参数位置 0 是对象 id 所在的位置，那么您可以测试是否 args[0 ] 匹配对象 ID。

从本质上讲，如果出现以下情况，则作业与对象 ID 相关联:job_class == class.name && job_queue == @queue && job_args[OBJECT_ID_ARGS_INDEX].to_i == object_id

• 排队作业:要查找所有排队作业，您需要收集所有 redis带有名为 queue:#{@queue} 键的条目，其中 @queue 是您的 worker 类(Class)正在使用的队列的名称。相应修改如果您使用多个队列，则循环多个队列一个特定的 worker 类(Class)。 Resque.redis.lrange("queue:#{@queue}",0,-1)
• 失败的作业:要找到所有排队的作业，您需要收集所有 redis带有名为 failed 键的条目(除非您使用多个故障队列或默认设置以外的一些)。 Resque.redis.lrange("failed",0,-1)
• 工作职位:要查找所有工作职位，您可以使用 Resque.workers其中包含所有工作人员和正在运行的作业的数组。 Resque.workers.map(&:job)
• 工作:上述每个列表中的每个工作都是一个编码哈希。你可以使用 Resque.decode(job) 将作业解码为 ruby​​ 哈希。
• Class 和 args:对于 queued jobs，class 和 args 键是 job[ “类”]和作业["args"]。对于失败和工作，这些是job["payload"]["class"]和job["payload"][ “参数”]。
• 队列:对于找到的每个失败和工作作业，队列将是job["queue"]。在测试对象 ID 的 args 列表之前，您只需要匹配 class 和 queue 的作业。您的排队作业列表将仅限于您收集的队列。

下面是示例工作类和模型方法，用于查找(和删除)与示例模型对象(报告)关联的作业。

resque/redis 支持的作业的报告处理 worker 类

class ProcessReportWorker
  # queue name
  @queue = :report_processing
  # tell the worker class where the report id is in the arguments list
  REPORT_ID_ARGS_INDEX = 0 

  # <snip> rest of class, not needed here for this answer

  # find jobs methods - find by report id (report is the 'associated' object)

  def self.find_queued_jobs_by_report_id report_id
    queued_jobs(@queue).select do |job|
      is_job_for_report? :queued, job, report_id
    end
  end

  def self.find_failed_jobs_by_report_id report_id
    failed_jobs.select do |job|
      is_job_for_report? :failed, job, report_id
    end
  end

  def self.find_working_jobs_by_report_id report_id
    working_jobs.select do |worker,job|
      is_job_for_report? :working, job, report_id
    end
  end

  # association test method - determine if this job is associated      

  def self.is_job_for_report? state, job, report_id
    attributes = job_attributes(state, job)
    attributes[:klass] == self.name && 
      attributes[:queue] == @queue && 
        attributes[:args][REPORT_ID_ARGS_INDEX].to_i == report_id
  end

  # remove jobs methods

  def self.remove_failed_jobs_by_report_id report_id
    find_failed_jobs_by_report_id(report_id).each do |job|
      Resque::Failure.remove(job["index"]) 
    end
  end

  def self.remove_queued_jobs_by_report_id report_id
    find_queued_jobs_by_report_id(report_id).each do |job|
      Resque::Job.destroy(@queue,job["class"],*job["args"])
    end
  end

  # reusable methods - these methods could go elsewhere and be reusable across worker classes

  # job attributes method

  def self.job_attributes(state, job)
    if state == :queued && job["args"].present?
      args = job["args"]
      klass = job["class"]
    elsif job["payload"] && job["payload"]["args"].present?
      args = job["payload"]["args"]
      klass = job["payload"]["class"] 
    else
      return {args: nil, klass: nil, queue: nil}
    end
    {args: args, klass: klass, queue: job["queue"]}
  end

  # jobs list methods

  def self.queued_jobs queue
    Resque.redis.lrange("queue:#{queue}", 0, -1)
      .collect do |job| 
        job = Resque.decode(job)
        job["queue"] = queue # for consistency only
        job
      end
  end

  def self.failed_jobs
    Resque.redis.lrange("failed", 0, -1)
      .each_with_index.collect do |job,index|
        job = Resque.decode(job)
        job["index"] = index # required if removing
        job
      end
  end

  def self.working_jobs
    Resque.workers.zip(Resque.workers.map(&:job))
      .reject { |w, j| w.idle? || j['queue'].nil? }
  end

end

那么 Report 模型的使用示例就变成了

class Report < ActiveRecord::Base
  before_destroy :check_if_working_jobs, :remove_queued_and_failed_jobs

  def check_if_working_jobs
    # find all working jobs related to this report object
    working_jobs = ProcessReportWorker.find_working_jobs_by_report_id(self.id) 
    return false unless working_jobs.empty?
  end

  def remove_queued_and_failed_jobs
    # find all jobs related to this report object
    queued_jobs = ProcessReportWorker.find_queued_jobs_by_report_id(self.id) 
    failed_jobs = ProcessReportWorker.find_failed_jobs_by_report_id(self.id) 

    # extra code and conditionals here for example only as all that is really 
    # needed is to call the remove methods without first finding or checking

    unless queued_jobs.empty?
      ProcessReportWorker.remove_queued_jobs_by_report_id(self.id)
    end

    unless failed_jobs.empty?
      ProcessReportWorker.remove_failed_jobs_by_report_id(self.id)
    end

  end
end

如果您为工作类使用多个队列或者您有多个故障队列，则需要修改解决方案。此外，还使用了 redis 故障后端。如果使用不同的故障后端，则可能需要进行更改。