ruby-on-rails - 快速将多个项目(1000/秒)添加到 sidekiq 队列？

Question

我知道 sidekiq 有一个 push_bulk 选项，但我目前受到 redis 延迟的限制，所以通过 push_bulk 传递多个项目仍然不够快(只有大约 50/s)。

我试过像这样增加 redis 连接的数量:

redis_conn = proc {
  Redis.new({ :url => Rails.configuration.redis.url })
}

Sidekiq.configure_client do |config|
  Sidekiq.configure_client do |config|
    config.redis = ConnectionPool.new(size: 50, &redis_conn)
  end
  config.client_middleware do |chain|
    chain.add Sidekiq::Status::ClientMiddleware
  end
end

然后启动单独的线程 (Thread.new) 以实际对各种对象执行 perform_async。有趣的是，任何不是第一个线程的线程都不会被扔进 sidekiq 队列，就好像它们被完全忽略了一样。

有谁知道更好的方法吗？

编辑:这是我正在尝试的 push_bulk 方法，实际上速度较慢:

  user_ids = User.need_scraping.pluck(:id)
  bar = ProgressBar.new(user_ids.count)
  user_ids.in_groups_of(10000, false).each do |user_id_group|
    Sidekiq::Client.push_bulk(
      'args'  => user_id_group.map{ |user_id| [user_id] },
      'class' => ScrapeUser,
      'queue' => 'scrape_user',
      'retry' => true
    )
  end

谢谢!

Answer 1

您确实想使用 push_bulk。您受到延迟/往返时间的限制，无法将元素写入支持 sidekiq 的 redis 队列。

当您真正应该删除额外的网络往返时，您正在使用多个线程/连接来克服缓慢的网络。

假设您正在尝试将 20k 个采用 user_id 的 UserWorker 作业加入队列:

您可以通过以下方式对单个作业进行排队:

UserWorker.perform_async(user_id)

...映射到:

Sidekiq::Client.push('class' => UserWorker, 'args' => [user_id] )

所以 20k user_ids 的 push_bulk 版本是:

# This example takes 20k user_ids in an array, chunks them into groups of 1000 ids,
# and batch sends them to redis as a group.

User.need_scraping.select('id').find_in_batches do |user_group|

  sidekiq_items = user_group.map {|user| { 'class' => UserWorker, 'args' => [user.id] } }
  Sidekiq::Client.push_bulk(sidekiq_items)
end

这将 20k 个 redis 调用变成了 20 个 redis 调用，平均往返时间为 5 毫秒(乐观)，即 1 秒对 100 秒。您的里程可能会有所不同。

编辑:评论者似乎对 Sidekiq/Redis 客户端批量排队数据的行为感到困惑。

Sidekiq::Client.push_bulk() 方法接受一组要排队的作业。它将这些转换为 Sidekiq 作业有效负载哈希，然后调用 SideKiq::Client.raw_push() 将这些有效负载传送到 redis。查看来源:https://github.com/mperham/sidekiq/blob/master/lib/sidekiq/client.rb#L158

SideKiq::Client.raw_push() 获取 Sidekiq 哈希负载列表，将它们转换为 JSON，然后执行结合两个 redis 命令的 redis MULTI 命令。首先，它将目标队列添加到事件队列列表 (redis SADD)，然后将所有作业有效负载推送到目标队列 redis 列表对象 (redis LPUSH) ).这是一个单独的 redis 命令，在一个单独的 redis 原子组中一起执行。

如果这仍然很慢，您可能有其他问题(网络速度慢、redis 服务器过载等)。