如何提升Ruby on Rails大批次数据校验环节的性能？

问题背景 #

基于Ruby on Rails、Postgres、Redis、Sidekiq、dry-validation的批量数据导入流程，处理最多30万行数据时，校验环节耗时超一天，核心瓶颈在于sku+name的唯一性校验：

• 原方案每行单独执行Product.where(sku: values[:sku]).where.not(name: values[:name]).exists?，导致数据库请求量暴增、负载极高；
• 尝试将全量数据pluck到数组后遍历对比，因数组查找效率低，耗时更长。

优化建议 #

1. 数据库索引优化（最基础且高效） #

针对sku+name的校验规则，创建复合索引，让数据库查询直接命中索引，避免全表扫描：

# 迁移文件中添加复合索引
add_index :products, [:sku, :name]

该索引会直接覆盖where(sku: x).where.not(name: y).exists?的查询逻辑，单条查询耗时从毫秒级降至微秒级。

2. 批量预查询+内存哈希校验（减少DB请求） #

不要每行单独查DB，而是一次性获取所有待校验sku对应的name集合，存入内存哈希结构（O(1)查找）：

# 提前收集所有待校验行的sku
incoming_skus = rows.map { |row| row[:sku] }.uniq

# 批量查询现有sku对应的name集合，转成哈希：sku => 对应的name集合
existing_sku_names = Product.where(sku: incoming_skus)
                           .pluck(:sku, :name)
                           .group_by(&:first)
                           .transform_values { |pairs| pairs.map(&:second).to_set }

# 在dry-validation规则中使用哈希校验
rule(:sku, :name) do
  if existing_sku_names.key?(values[:sku]) && !existing_sku_names[values[:sku]].include?(values[:name])
    key.failure("该SKU已存在且名称不匹配")
  end
end

此方案将DB请求从30万次减少到1次，校验逻辑转为内存操作，性能提升数量级。

3. Sidekiq并行拆分任务（利用多核CPU） #

将30万行数据拆分为小批次（如每1000行一批），通过Sidekiq并行处理，充分利用服务器多核资源：

# 拆分任务到Sidekiq
rows.each_slice(1000) do |batch|
  ProductValidationWorker.perform_async(batch, existing_sku_names)
end

# 对应的Worker
class ProductValidationWorker
  include Sidekiq::Worker

  def perform(batch, existing_sku_names)
    validator = ProductImportValidator.new(context: { existing_sku_names: existing_sku_names })
    batch.each do |row|
      result = validator.call(row)
      set_status(row, result) # 标记有效/无效
    end
  end
end

注意通过Sidekiq配置控制并发数（如concurrency: 10），避免数据库瞬间过载。

4. dry-validation上下文复用（减少重复计算） #

将预查询的existing_sku_names作为校验上下文传递，避免每个rule重复加载数据：

class ProductImportValidator < Dry::Validation::Contract
  params do
    required(:sku).filled(:string)
    required(:name).filled(:string)
    # 其他字段定义
  end

  rule(:sku, :name) do
    existing_sku_names = context[:existing_sku_names]
    next unless existing_sku_names.key?(values[:sku])

    key.failure("该SKU已存在且名称不匹配") unless existing_sku_names[values[:sku]].include?(values[:name])
  end
end

# 使用时传入上下文
validator = ProductImportValidator.new(context: { existing_sku_names: existing_sku_names })

5. Postgres临时表批量校验（极致性能） #

将待校验数据导入Postgres临时表，通过SQL批量找出冲突行，利用数据库的批量处理能力：

# 创建临时表
ActiveRecord::Base.connection.execute(<<~SQL)
  CREATE TEMP TABLE temp_import_rows (sku TEXT, name TEXT);
SQL

# 批量导入待校验数据（可使用COPY命令提升速度）
rows.each do |row|
  ActiveRecord::Base.connection.execute(<<~SQL)
    INSERT INTO temp_import_rows (sku, name) VALUES ('#{row[:sku]}', '#{row[:name]}');
  SQL
end

# 查询所有冲突行
conflicts = ActiveRecord::Base.connection.execute(<<~SQL)
  SELECT t.sku, t.name 
  FROM temp_import_rows t
  JOIN products p ON t.sku = p.sku AND t.name != p.name;
SQL

# 标记冲突行无效
conflict_set = conflicts.to_set { |row| [row['sku'], row['name']] }
rows.each do |row|
  if conflict_set.include?([row[:sku], row[:name]])
    set_status(row, :invalid)
  else
    set_status(row, :valid)
  end
end

此方案将校验逻辑转移到数据库层面，处理30万行数据的耗时可压缩至分钟级。

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