如何获取Sidekiq Worker的并发运行上限以高效批量处理20万用户数据

• 数据库共存储200000个用户数据，需要遍历每条用户记录执行对应业务逻辑
• 实现思路为编写Rake任务遍历用户，核心业务逻辑放到Sidekiq worker中执行
• 最初的规划：如果能获取到Sidekiq可同时运行的worker数量上限（例如上限为50000），就把用户划分为对应数量的批次，逐批次为worker分配任务
• 当前编写的Rake任务代码：

namespace :users do
  task data: :environment do
   confirmed_users = User.where('confirmed_at IS NOT NULL').where('id <= 50000')
   confirmed_users.each do |user|
     MyWorker.perform_async(user.id)
   end

   confirmed_users = User.where('confirmed_at IS NOT NULL').where('id > 50000 and id <= 100000')
   confirmed_users.each do |user|
     MyWorker.perform_async(user.id)
   end

   confirmed_users = User.where('confirmed_at IS NOT NULL').where('id > 100000 and id <= 150000')
   confirmed_users.each do |user|
     MyWorker.perform_async(user.id)
   end

   confirmed_users = User.where('confirmed_at IS NOT NULL').where('id > 200000')
   confirmed_users.each do |user|
     MyWorker.perform_async(user.id)
   end
 end
end

待解决的核心疑问 #

1. 如何获取Sidekiq的并发运行上限，实现动态生成用户处理批次
2. 当前的实现方式是否能最大限度缩短处理耗时，是否存在更高效的全量用户记录处理方案

关于Sidekiq并发上限的认知与获取 #

• 首先需要纠正一个认知偏差：Sidekiq的并发上限指的是单时刻可同时执行的worker数量，不是队列可容纳的任务总数。队列本身基于Redis实现，没有任务数上限，哪怕堆放几百万任务也不会占用运行资源，只有正在执行的任务会消耗服务器CPU、内存、数据库连接资源。
• Sidekiq的并发值可以直接读取配置获取，不需要硬编码：读取的参数为Sidekiq.options[:concurrency]，这个值是单个Sidekiq进程开启的工作线程数，生产环境通常配置为5-25（受Ruby全局解释锁限制，过高的线程数不会提升性能，反而会造成资源竞争）。如果部署了多个Sidekiq进程，总并发值为所有进程的并发配置之和。你之前设想的“并发上限50000”完全不符合实际运行逻辑。

当前实现的问题 #

• 分段的写法没有实际作用：无论你把用户按id拆成多少个区间遍历，循环中调用的perform_async都会立刻把任务推入Sidekiq队列，Rake任务本身几秒钟就会执行完成，后续任务的执行节奏完全由Sidekiq的并发配置控制，分段逻辑和一次性全量推任务没有任何区别，根本达不到“按并发值分批次”的目的。
• 逻辑存在明显漏洞：总用户量为200000，最后一个查询条件为id > 200000，永远不会匹配到任何数据，直接跳过了150001-200000区间的用户，会造成约25%的用户没有被处理。
• 全量遍历有内存溢出风险：直接用each遍历查询结果，会一次性把所有匹配的用户对象加载到内存中，当用户量达到几十万级别时，会直接超出Rake任务的内存限制。

更高效的优化实现 #

• 移除手动分段逻辑，直接批量推送任务：利用Sidekiq原生的perform_bulk能力，把任务按1000个为一组批量写入Redis，相比循环调用perform_async（每个任务单独发一次Redis请求），推送20万任务的耗时可以从几十分钟压缩到几秒。参考实现：

task data: :environment do
  # 仅查询需要的用户id，不加载完整用户对象，减少内存占用
  confirmed_user_ids = User.where.not(confirmed_at: nil).pluck(:id)
  # 按1000个任务为一组批量推送
  confirmed_user_ids.each_slice(1000) do |id_batch|
    MyWorker.perform_bulk(id_batch.map { |uid| [uid] })
  end
end

• 不需要手动按并发值控制流量：Sidekiq本身的队列机制已经天然实现了流量控制，哪怕队列里堆20万任务，Sidekiq也只会按照配置的并发数同步执行，不会打满服务器资源。如果需要限制全量任务的数据库压力，可以通过Sidekiq的速率限制、队列优先级配置实现，不需要在Rake任务中写死分段逻辑。
• 需要跟踪全量任务进度的话，可以使用Sidekiq内置的Batch功能，创建任务批次后可以实时查看已完成数、失败数、总进度，比手动写分段逻辑方便很多。

内容的提问来源于stack exchange，提问作者honey