使用 delay_job 轮询答案

【问题标题】：polling with delayed_job使用 delay_job 轮询
【发布时间】：2011-07-31 17:18:15
【问题描述】：

我有一个通常需要几秒钟才能完成的过程，所以我尝试使用delayed_job 来异步处理它。工作本身运行良好，我的问题是如何对工作进行轮询以了解它是否已完成。

我可以通过简单地将其分配给一个变量来从delayed_job 中获取一个id：

job = Available.delay.dosomething(:var => 1234)

+------+----------+----------+------------+------------+-------------+-----------+-----------+-----------+------------+-------------+
| id   | priority | attempts | handler    | last_error | run_at      | locked_at | failed_at | locked_by | created_at | updated_at  |
+------+----------+----------+------------+------------+-------------+-----------+-----------+-----------+------------+-------------+
| 4037 | 0        | 0        | --- !ru... |            | 2011-04-... |           |           |           | 2011-04... | 2011-04-... |
+------+----------+----------+------------+------------+-------------+-----------+-----------+-----------+------------+-------------+

但一旦完成作业，它就会将其删除并搜索已完成的记录会返回错误：

@job=Delayed::Job.find(4037)

ActiveRecord::RecordNotFound: Couldn't find Delayed::Backend::ActiveRecord::Job with ID=4037

@job= Delayed::Job.exists?(params[:id])

我是否应该费心改变这一点，或者推迟删除完整记录？我不知道我还能如何获得它的状态通知。还是轮询死记录作为完成证明可以吗？还有其他人面临类似的情况吗？

【问题讨论】：

我面临的另一个问题或障碍是，我正在抵消这份工作，因为否则它会占用我的服务器。我询问数据库是否存在某些带有日期的数据，如果它们不存在或已过时，我会获取新数据，但是作为一项单独的工作，我使用 AJAX 并轮询直到它完成......然后运行再次查询新数据。让它变得更快，但也更复杂。
使用 resque 和 redis 之类的东西并基本上缓存返回的对象，这样可以节省更多往返数据库的时间，并使轮询速度更快吗？我从来没有接触过 redis 或 resque，所以我想我会把它扔在那里。

标签： ruby-on-rails ruby-on-rails-3 backgroundworker push-notification delayed-job

【解决方案1】：

我建议，如果获取作业已完成的通知很重要，那么编写一个自定义作业对象并将排队，而不是依赖于调用 @ 时排队的默认作业987654321@。创建一个类似的对象：

class DoSomethingAvailableJob

  attr_accessor options

  def initialize(options = {})
    @options = options
  end

  def perform
    Available.dosomething(@options)
    # Do some sort of notification here
    # ...
  end
end

并将其加入队列：

Delayed::Job.enqueue DoSomethingAvailableJob.new(:var => 1234)

【讨论】：

是的，我知道我可以在成功或错误回调之后编写，但是该作业在成功时在数据库中创建/更新约 50-100 条记录，因为我正在轮询它，所以我寻找要查找的东西而不是能够推动某些东西。如果这有意义的话。
我正在使用 ajax 请求来确定它是否已完成，这就是为什么我无法推送它...除非我对套接字做了一些疯狂的事情。
我在想更多的“更新某处的标志”，您的 AJAX 请求只能轮询。

【解决方案2】：

您的应用程序中的delayed_jobs 表仅用于提供正在运行和排队的作业的状态。它不是一个持久表，出于性能原因，它确实应该尽可能小。这就是作业在完成后立即被删除的原因。

相反，您应该向您的 Available 模型添加表示工作已完成的字段。由于我通常对处理作业需要多长时间感兴趣，因此我添加了 start_time 和 end_time 字段。然后我的dosomething 方法看起来像这样：

def self.dosomething(model_id)

 model = Model.find(model_id)

  begin
    model.start!

    # do some long work ...

    rescue Exception => e
      # ...
    ensure
      model.finish!
  end
end

开始！并完成！方法只是记录当前时间并保存模型。然后我会有一个completed? 方法，您的 AJAX 可以轮询该方法以查看作业是否完成。

def completed?
  return true if start_time and end_time
  return false
end

有很多方法可以做到这一点，但我觉得这个方法很简单，对我来说效果很好。

【讨论】：

【解决方案3】：

让我们从 API 开始。我想要类似下面的东西。

@available.working? # => true or false, so we know it's running
@available.finished? # => true or false, so we know it's finished (already ran)

现在让我们来写作业吧。

class AwesomeJob < Struct.new(:options)

  def perform
    do_something_with(options[:var])
  end

end

到目前为止一切顺利。我们有一份工作。现在让我们编写将其排入队列的逻辑。由于 Available 是负责这项工作的模型，我们来教它如何开始这项工作。

class Available < ActiveRecord::Base

  def start_working!
    Delayed::Job.enqueue(AwesomeJob.new(options))
  end

  def working?
    # not sure what to put here yet
  end

  def finished?
    # not sure what to put here yet
  end

end

那么我们如何知道工作是否有效？有几种方法，但在 Rails 中，当我的模型创建某些东西时，它通常与那个东西相关联。我们如何关联？在数据库中使用 id。让我们在可用模型上添加一个job_id。

当我们在做的时候，我们怎么知道这个工作是因为它已经完成了，还是因为它还没有开始？一种方法是实际检查工作实际做了什么。如果它创建了一个文件，请检查文件是否存在。如果它计算了一个值，请检查结果是否已写入。但是，有些工作并不容易检查，因为他们的工作可能没有明确的可验证结果。对于这种情况，您可以在模型中使用标志或时间戳。假设这是我们的情况，让我们添加一个job_finished_at 时间戳来区分尚未运行的作业和已经完成的作业。

class AddJobIdToAvailable < ActiveRecord::Migration
  def self.up
    add_column :available, :job_id, :integer
    add_column :available, :job_finished_at, :datetime
  end

  def self.down
    remove_column :available, :job_id
    remove_column :available, :job_finished_at
  end
end

好的。所以现在让我们通过修改start_working! 方法，将Available 与它的工作相关联。

def start_working!
  job = Delayed::Job.enqueue(AwesomeJob.new(options))
  update_attribute(:job_id, job.id)
end

太好了。在这一点上，我本可以写 belongs_to :job，但我们并不真的需要它。

所以现在我们知道如何编写working? 方法了，非常简单。

def working?
  job_id.present?
end

但是我们如何标记工作已完成？没有人知道工作比工作本身做得更好。因此，让我们将available_id 传递给作业（作为选项之一）并在作业中使用它。为此我们需要修改start_working! 方法来传递id。

def start_working!
  job = Delayed::Job.enqueue(AwesomeJob.new(options.merge(:available_id => id))
  update_attribute(:job_id, job.id)
end

我们应该将逻辑添加到作业中，以便在完成时更新我们的 job_finished_at 时间戳。

class AwesomeJob < Struct.new(:options)

  def perform
    available = Available.find(options[:available_id])
    do_something_with(options[:var])

    # Depending on whether you consider an error'ed job to be finished
    # you may want to put this under an ensure. This way the job
    # will be deemed finished even if it error'ed out.
    available.update_attribute(:job_finished_at, Time.current)
  end

end

有了这段代码，我们就知道如何编写finished? 方法了。

def finished?
  job_finished_at.present?
end

我们完成了。现在我们可以简单地对@available.working? 和@available.finished? 进行轮询。此外，您还可以通过检查@available.job_id 来方便地了解为您的Available 创建的确切工作。您可以通过说belongs_to :job 轻松地将其转化为真正的关联。

【讨论】：

【解决方案4】：

我认为最好的方法是使用延迟作业中可用的回调。这些是： :success、:error 和 :after。所以你可以在你的模型中加入一些代码：

class ToBeDelayed
  def perform
    # do something
  end

  def after(job)
    # do something
  end
end

因为如果你坚持使用 obj.delayed.method，那么你将不得不修改 Delayed::PerformableMethod 并在那里添加 after 方法。恕我直言，这比轮询某些甚至可能是后端特定的值（例如 ActiveRecord 与 Mongoid）要好得多。

【讨论】：

回调工作得很好。现在我可以在工作完成或失败时执行我自己的逻辑。谢谢。
这些服务器端回调对试图确定作业何时完成的客户端有何帮助？
after 回调可以：1.设置一些标志，2.客户端将轮询该标志是否变为真，或者使用消息系统作为 faye 通过 websocket 向客户端发出进程已完成的信号.
啊，我理解错了——你的意思是使用回调而不是轮询

【解决方案5】：

完成此操作的最简单方法是将轮询操作更改为类似于以下内容：

def poll
  @job = Delayed::Job.find_by_id(params[:job_id])

  if @job.nil?
    # The job has completed and is no longer in the database.
  else
    if @job.last_error.nil?
      # The job is still in the queue and has not been run.
    else
      # The job has encountered an error.
    end
  end
end

为什么会这样？当Delayed::Job 从队列中运行作业时，如果成功，它将从数据库中删除它。如果作业失败，则记录保留在队列中，稍后再次运行，并将last_error 属性设置为遇到的错误。使用上述两个功能，您可以检查已删除的记录是否成功。

上述方法的好处是：

您获得了您在原始帖子中寻找的投票效果
使用简单的逻辑分支，如果在处理作业中出现错误，您可以向用户提供反馈

您可以通过执行以下操作将此功能封装在模型方法中：

# Include this in your initializers somewhere
class Queue < Delayed::Job
  def self.status(id)
    self.find_by_id(id).nil? ? "success" : (job.last_error.nil? ? "queued" : "failure")
  end
end

# Use this method in your poll method like so:
def poll
    status = Queue.status(params[:id])
    if status == "success"
      # Success, notify the user!
    elsif status == "failure"
      # Failure, notify the user!
    end
end

【讨论】：

非常简单，看似有效。这种技术有什么陷阱或缺点？
唯一真正的缺点是检查已删除记录似乎有点“hackish”，而不是使用 Rails 中首选的回调/观察者系统。
你会将job_id存储在哪里，这样页面刷新就不会丢失它？

【解决方案6】：

我最终使用了 Delayed_Job 与 after(job) 回调的组合，该回调使用与创建的作业相同的 ID 填充 memcached 对象。通过这种方式，我最大限度地减少了访问数据库询问作业状态的次数，而不是轮询 memcached 对象。它包含完成作业所需的全部对象，所以我什至没有往返请求。我从 github 人的一篇文章中得到了这个想法，他们做了几乎同样的事情。

https://github.com/blog/467-smart-js-polling

并使用jquery插件进行轮询，轮询频率较低，并在重试一定次数后放弃

https://github.com/jeremyw/jquery-smart-poll

看起来效果不错。

 def after(job)
    prices = Room.prices.where("space_id = ? AND bookdate BETWEEN ? AND ?", space_id.to_i, date_from, date_to).to_a
    Rails.cache.fetch(job.id) do
      bed = Bed.new(:space_id => space_id, :date_from => date_from, :date_to => date_to, :prices => prices)
    end
  end

【讨论】：