延迟作业的resque-status和resque-scheduler

Question

在前面的代码中，我对延迟作业使用了resque-scheduler：

Resque.enqueue_in(options[:delay].seconds, self, context)

现在我想包括resque-status来做这项工作，但不知道他们如何在一起工作。最新的resque-status源码支持scheduler，如源码：

https://github.com/quirkey/resque-status/blob/master/lib/resque/plugins/status.rb

# Wrapper API to forward a Resque::Job creation API call into a Resque::Plugins::Status call.
# This is needed to be used with resque scheduler
# http://github.com/bvandenbos/resque-scheduler
def scheduled(queue, klass, *args)
  self.enqueue_to(queue, self, *args)
end

结束

但是我不确定如何使用它。我应该只调用SampleJob.scheduled(queue，myclass，:delay => delay)而不是SampleJob.create(options)吗？

======================================================================

此外，还支持resque-status (和其他自定义作业)：

https://github.com/bvandenbos/resque-scheduler

一些Resque扩展，如resque-status，使用具有稍微不同的API签名的自定义作业类。Resque-scheduler并不是试图支持所有现有的和未来的自定义作业类，相反，它支持一个调度标志，这样你就可以扩展你的自定义类，使其支持调度的作业。

假设我们有一个名为FakeLeaderboard的JobWithStatus类

class FakeLeaderboard < Resque::JobWithStatus
  def perform
    # do something and keep track of the status
  end
end

然后是一个时间表：

create_fake_leaderboards:
  cron: "30 6 * * 1"
  queue: scoring
  custom_job_class: FakeLeaderboard
  args:
  rails_env: demo
  description: "This job will auto-create leaderboards for our online demo and the status will update as the worker makes progress"

但它似乎只适用于重复出现的作业。我可以找到cron的参数，但不能找到延迟。那么我如何使用它来处理延迟的作业呢？

谢谢!

Answer 1

我也遇到了同样的问题，我自己解决了这个问题，实现了一个模块，它为“状态”作业提供了一个运行器。

module Resque # :nodoc:

  # Module to include in your worker class to get resque-status
  # and resque-scheduler integration
  module ScheduledJobWithStatus

    extend ActiveSupport::Concern

    included do
      # Include status functionalities
      include Resque::Plugins::Status
    end

    # :nodoc:
    module ClassMethods

      # This method will use a custom worker class to enqueue jobs
      # with resque-scheduler plugin with status support
      def enqueue_at(timestamp, *args)
        class_name = self.to_s # store class name since plain class object are not "serializable"
        Resque.enqueue_at(timestamp, JobWithStatusRunner, class_name, *args)
      end

      # Identical to enqueue_at but takes number_of_seconds_from_now
      # instead of a timestamp.
      def enqueue_in(number_of_seconds_from_now, *args)
        enqueue_at(Time.now + number_of_seconds_from_now, *args)
      end

    end
  end

  # Wrapper worker for enqueuing
  class JobWithStatusRunner
    # default queue for scheduling jobs with status
    @queue = :delayed

    # Receive jobs from {Resque::ScheduledJobWithStatus} queue them in Resque
    # with support for status informations
    def self.perform(status_klass, *args)
      # Retrieve original worker class
      klass = status_klass.to_s.constantize
      # Check if supports status jobs
      unless klass.included_modules.include? Resque::Plugins::Status
        Rails.logger.warn("Class #{klass} doesn't support jobs with status")
        return false
      end
      Rails.logger.debug("Enqueing jobs #{klass} with arguments #{args}")
      klass.create(*args)
    rescue NameError
      Rails.logger.error("Unable to enqueue jobs for class #{status_klass} with args #{args}")
      false
    end
  end

end

通过这种方式，您可以使用以下简单语法将作业排入队列：

# simple worker class
class SleepJob

  # This provides integrations with both resque-status and resque-scheduler
  include Resque::ScheduledJobWithStatus

  # Method triggered by resque
  def perform
    total = (options['length'] || 60).to_i
    1.upto(total) { |i| at(i, total, "At #{i} of #{total}"); sleep(1) }
  end

end

# run the job delayed
SleepJob.enqueue_in(5.minutes)
# or
SleepJob.enqueue_at(5.minutes.from_now)

只需将模块放在resque初始化器或lib文件夹中即可。在后一种情况下，记得在某个地方需要它。

Answer 2

每当resque-scheduler创建供工作人员使用的作业时，它都会在所提供的类上调用scheduled方法。它还可以很好地将队列名和类名作为字符串传递，因此您可以创建一个类级方法来处理计划的作业创建。

虽然Fabio的答案将解决问题，但它在回答您的特定问题时有点过度设计。假设您有一个所有resque-status工作者都继承的JobWithStatus类，您只需向其添加scheduled方法，它就可以像这样使用resque-scheduler：

class JobWithStatus
  include Resque::Plugins::Status

  def self.scheduled(queue, klass, *args)
    Resque.constantize(klass).create(*args)
  end
end