使用 Sidekiq 完成大任务或多个小任务

Question

我正在编写一个工人来将很多用户添加到一个组中。我想知道是否最好运行一个拥有所有用户的大型任务，或者像 100 个用户一样批量运行，或者每个任务一个接一个地运行。

这里是我的代码

class AddUsersToGroupWorker
  include Sidekiq::Worker
  sidekiq_options :queue => :group_utility

  def perform(store_id, group_id, user_ids_to_add)
    begin
      store = Store.find store_id
      group = Group.find group_id
    rescue ActiveRecord::RecordNotFound => e
      Airbrake.notify e
      return
    end

    users_to_process = store.users.where(id: user_ids_to_add)
                                  .where.not(id: group.user_ids)
    group.users += users_to_process

    users_to_process.map(&:id).each do |user_to_process_id|
      UpdateLastUpdatesForUserWorker.perform_async store.id, user_to_process_id
    end
  end
end 

也许在我的方法中有这样的东西会更好：

def add_users
    users_to_process = store.users.where(id: user_ids_to_add)
                                  .where.not(id: group.user_ids)

    users_to_process.map(&:id).each do |user_to_process_id|
      AddUserToGroupWorker.perform_async group_id, user_to_process_id
      UpdateLastUpdatesForUserWorker.perform_async store.id, user_to_process_id
    end
end

但是有很多find 请求。你怎么看？

如果需要，我有一个 sidekig pro 许可证（例如批量）。

Answer 1

这是我的想法。

1.执行单个 SQL 查询而不是 N 个查询

这一行：group.users += users_to_process 可能会产生 N 个 SQL 查询（其中 N 是 users_to_process.count）。我假设您在用户和组之间有多对多连接（使用user_groups 连接表/模型），所以您应该使用一些Mass inserting data technique：

users_to_process_ids = store.users.where(id: user_ids_to_add)
                         .where.not(id: group.user_ids)
                         .pluck(:id)
sql_values = users_to_process_ids.map{|i| "(#{i.to_i}, #{group.id.to_i}, NOW(), NOW())"}
Group.connection.execute("
  INSERT INTO groups_users (user_id, group_id, created_at, updated_at)
  VALUES #{sql_values.join(",")}
")

是的，它是原始 SQL。而且它很快。

2。用户 pluck(:id) 而不是 map(&:id)

pluck 更快，因为：

• 它将仅选择“id”列，因此从 DB 传输的数据较少
• 更重要的是，它不会为每个原始数据创建 ActiveRecord 对象

执行 SQL 很便宜。创建 Ruby 对象非常昂贵。

3。使用水平并行化而不是垂直并行化

我这里的意思是，如果你需要为十几条记录做顺序任务A -> B -> C，有两种主要的工作拆分方式：

• 垂直分割。 AWorker 做 A(1), A(2), A(3); BWorker 做B(1) 等； CWorker 完成所有 C(i) 工作；
• 水平分割。 UniversalWorker 是 A(1)+B(1)+C(1)。

使用后一种（水平）方式。

这是来自经验的陈述，而不是从某种理论的角度来看（两种方式都可行）。

为什么要这样做？

• 当您使用垂直分段时，您可能会在将工作从一名工作人员传递给另一名工作人员时出错。喜欢such kind of errors。如果你遇到这样的错误，你会拔掉头发，因为它们不是持久的，也不容易重现。有时它们会发生，有时它们不会。是否可以编写一个代码，将工作沿着链向下传递而不会出错？是的。不过最好keep it simple。
• 假设您的服务器处于静止状态。然后突然新的工作来了。你的B 和C 工作人员只会浪费内存，而你的A 工作人员会完成这项工作。然后你的A 和C 将浪费内存，而B 正在工作。等等。如果您进行水平分割，您的资源消耗会自行解决。

将该建议应用于您的具体案例：对于初学者，不要在另一个异步任务中调用 perform_async。

4.分批处理

回答您最初的问题 - 是的，分批处理。创建和管理异步任务本身会占用一些资源，因此无需创建太多。

---

TL;DR所以最后，您的代码可能看起来像这样：

# model code

BATCH_SIZE = 100

def add_users
  users_to_process_ids = store.users.where(id: user_ids_to_add)
                           .where.not(id: group.user_ids)
                           .pluck(:id)
  # With 100,000 users performance of this query should be acceptable
  # to make it in a synchronous fasion
  sql_values = users_to_process_ids.map{|i| "(#{i.to_i}, #{group.id.to_i}, NOW(), NOW())"}
  Group.connection.execute("
    INSERT INTO groups_users (user_id, group_id, created_at, updated_at)
    VALUES #{sql_values.join(",")}
  ")

  users_to_process_ids.each_slice(BATCH_SIZE) do |batch|
    AddUserToGroupWorker.perform_async group_id, batch
  end
end

# add_user_to_group_worker.rb

def perform(group_id, user_ids_to_add)
  group = Group.find group_id

  # Do some heavy load with a batch as a whole
  # ...
  # ...
  # If nothing here is left, call UpdateLastUpdatesForUserWorker from the model instead

  user_ids_to_add.each do |id|
    # do it synchronously – we already parallelized the job
    # by splitting it in slices in the model above
    UpdateLastUpdatesForUserWorker.new.perform store.id, user_to_process_id
  end
end

Answer 2

没有灵丹妙药。这取决于您的目标和您的应用程序。要问自己的一般问题：

• 您可以将多少用户 ID 传递给工作人员？能过100吗？ 100万呢？
• 您的工人可以工作多长时间？它应该对工作时间有任何限制吗？他们能卡住吗？

对于大型应用程序，有必要将传递的参数拆分为更小的块，以避免创建长时间运行的作业。创建大量小型作业可以让您轻松扩展 - 您可以随时添加更多工人。

另外，为工作人员定义超时类型以停止处理卡住的工作人员可能是一个好主意。