使用特定于用户的队列进行后台处理？答案

【问题标题】：Background processing with user-specific queues?使用特定于用户的队列进行后台处理？
【发布时间】：2013-11-28 19:22:35
【问题描述】：

我一直在使用 Sidekiq 处理后台作业，但我发现它对于我的特定用例来说过于局限。

当用户创建帐户时，我们会从第三方服务导入他们的数据。该服务有一个速率限制，因此我可以将数十名工人投入其中以加快导入速度。

问题是我无法使用 Sidekiq 精细控制工人的数量。

我可以限制每个队列的工作人员数量，但这对我没有帮助。

例如，如果 10 个人创建了一个帐户，我必须对所有 10 个人的所有数据的 ENTIRE 导入进行速率限制，但我真正需要的是对每个单独的帐户进行速率限制。

实际上能够创建一个特定于用户的队列，然后限制每个队列的工作人员数量可能会起到作用。

有没有像 Sidekiq 这样可以更精细地控制工人数量的东西？

【问题讨论】：

标签： ruby-on-rails ruby background-process resque sidekiq

【解决方案1】：

我有类似的情况，我需要自己施加速率限制以避免 DDoSing 外部服务。我使用Sidetiq 来控制 Sidekiq 作业的排队。

使用 Sidetiq，您可以定期对 Sidekiq 作业进行排队。在您的情况下，您可以创建一个 Sidetiq 作业，例如，每小时将 10 个用户导入作业添加到 Sidekiq。

class UserJobScheduler
  include Sidekiq::Worker
  include Sidetiq::Schedulable

  recurrence { hourly.minute_of_hour(0) }

  sidekiq_options queue: :default

  def perform
    # find 10 unprocessed user ids, then queue them up
    User.where(unprocessed: true).limit(10).pluck(:id).each do |user_id|
      UserDataImporter.perform_async(user_id)
    end
  end
end

class UserDataImporter
  include Sidekiq::Worker

  sidekiq_options queue: :user_import, retry: false

  def perform(user_id)
    # import user data & mark as processed
  end
end

我在实际调用我需要速率限制的 API 的类中将重试设置为 false。这允许更精确地控制正在发送的请求数量。否则失败的请求将遵循 Sidekiq 的重试计划，这对于您调用的服务可能过于激进

然后，您可以在 UserJobScheduler 中调整每次运行添加的作业数量，以及将它们添加到 Sidekiq 队列的频率，并将重复设置降至低于您正在调用的 API 的速率限制。

【讨论】：