使用 Sidekiq 执行异步作业时处理陈旧数据答案

【问题标题】：Working with stale data when performing asynchronous jobs with Sidekiq使用 Sidekiq 执行异步作业时处理陈旧数据
【发布时间】：2015-05-14 23:01:12
【问题描述】：

为了异步处理事件并创建活动提要，我使用 Sidekiq 和 Ruby on Rails 的全局 ID。

这适用于大多数类型的活动，但其中一些活动需要的数据可能会在作业执行时发生变化。

这是一个完全虚构的例子：

class Movie < ActiveRecord::Base
  include Redis::Objects
  value :score # stores an integer in Redis

  has_many :likes

  def popular?
    likes.count > 1000
  end
end

每次更新电影时，Sidekiq 工作人员都会执行一项工作：

class MovieUpdatedWorker
  include Sidekiq::Worker

  def perform(global_id)
    movie = GlobalID::Locator.locate(global_id)
    MovieUpdatedActivity.create(movie: movie, score: movie.score) if movie.popular?
  end
end

现在，想象一下 Sidekiq 落后了，在它有机会执行其工作之前，电影的 score 在 Redis 中更新，一些用户不喜欢这部电影，popular 方法现在返回 false。

Sidekiq 最终使用更新的数据。

我正在寻找方法来安排作业，同时确保在执行作业时所需的数据不会改变。一些想法：

1/ 手动传入所有需要的数据并相应地调整工人：

MovieUpdatedWorker.perform_async(
  movie: self,
  score: score,
  likes_count: likes.count
)

这可行，但需要重新实现/复制所有依赖数据的方法，例如 score 和 popular?（想象一个应用程序远不止这两个/三个可移动部分）。

这也不能很好地扩展，因为序列化的对象可能会占用 Redis 中的大量空间。

2/ 在传递给工作人员的记录上存根一些方法：

MovieUpdatedWorker.perform_async(
  global_id,
  stubs: { score: score, popular?: popular? }
)

class MovieUpdatedWorker
  include Sidekiq::Worker

  def perform(global_id, stubs: {})
    movie = GlobalID::Locator.locate(global_id)

    # inspired by RSpec
    stubs.each do |message, return_value|
      movie.stub(message) { return_value }
    end

    MovieUpdatedActivity.create(movie: movie, score: movie.score) if movie.popular?
  end
end

这不是功能性的，但您可以想象处理实际记录的便利性，无需重新实现现有方法并处理实际数据。

您是否看到其他“冻结”对象数据并异步处理它们的策略？你觉得这两个怎么样？

【问题讨论】：

标签： ruby-on-rails ruby asynchronous sidekiq

【解决方案1】：

我不会说这些数据是陈旧的，因为您实际上会拥有它的最新版本，只是它不再流行。听起来您实际上想要的是旧版本。

如果您不希望数据发生更改，则需要以某种方式对其进行缓存。要么像你说的那样，直接将数据传递给作业，要么你可以在数据库中添加某种形式的数据版本，并传递对旧版本的引用。

我认为将你需要的数据传递给 Redis 是一种合理的方式。你可以只序列化你真正关心的属性，比如分数。

【讨论】：

你说得对，我修正了标题并改写了问题。谢谢！