ActiveJob处理大量用户ID的可行性及优化方案咨询

问题描述 #

我正在编写一个后台任务，用于处理18000个用户并为每位用户创建UserFlag。目前已获取用户ID，编写了如下代码：

class BatchProcessUsersJob < ApplicationJob
  queue_as :default

  def perform()
    user_ids = [
      # 18,000 ids
    ]

    UserFlag.add_flag_to_users(user_ids, UserFlag::PRODUCT_TYPE_2)
  end
end

UserFlag类中的相关方法：

def self.add_flag_to_users(user_ids, flag)
  users_with_flag = UserFlag.where(user_id: user_ids).where(flag: flag).pluck(:user_id)
  users_without_flags = user_ids - users_with_flag
  return if users_without_flags.empty?

  users_without_flags.uniq.each do |user_id|
    UserFlag.find_or_create_by(user_id: user_id, flag: flag)
  end
end

这种方式能否处理18000个用户？当前方案可能存在超时或其他问题，处理大量用户的方式不够高效，有没有更合适的实现方式？

当前方案的问题 #

• 循环调用find_or_create_by会发起数千次SQL查询，数据库压力大，极易触发超时
• 一次性加载18000个ID到内存，可能导致内存占用过高
• 缺乏异常处理机制，中间出错会导致部分用户未被处理

优化方案 #

1. 批量插入（减少SQL查询次数） #

利用Rails的批量插入能力，避免循环发起单条查询：

方案A：使用activerecord-import gem

先在Gemfile添加依赖：

gem 'activerecord-import'

执行bundle install后，修改方法：

def self.add_flag_to_users(user_ids, flag)
  existing_user_ids = UserFlag.where(user_id: user_ids, flag: flag).pluck(:user_id)
  target_user_ids = user_ids.uniq - existing_user_ids
  return if target_user_ids.empty?

  records = target_user_ids.map do |user_id|
    UserFlag.new(user_id: user_id, flag: flag)
  end
  UserFlag.import(records)
end

方案B：使用Rails原生insert_all（Rails 6+）

无需第三方gem，直接用内置方法：

def self.add_flag_to_users(user_ids, flag)
  existing_user_ids = UserFlag.where(user_id: user_ids, flag: flag).pluck(:user_id)
  target_user_ids = user_ids.uniq - existing_user_ids
  return if target_user_ids.empty?

  values = target_user_ids.map do |user_id|
    { user_id: user_id, flag: flag, created_at: Time.current, updated_at: Time.current }
  end
  UserFlag.insert_all(values)
end

2. 分批次处理（降低内存占用） #

如果一次性处理18000个ID内存压力大，可拆分批次处理：

修改Job代码：

class BatchProcessUsersJob < ApplicationJob
  queue_as :default

  def perform
    user_ids = [
      # 18,000 ids
    ]

    # 每次处理1000个，可根据数据库性能调整批次大小
    user_ids.each_slice(1000) do |batch_ids|
      UserFlag.add_flag_to_users(batch_ids, UserFlag::PRODUCT_TYPE_2)
    end
  end
end

配合批量插入方法，既控制内存占用，又减少数据库请求量。

3. 数据库原生SQL（极致性能） #

直接用SQL完成筛选和插入，跳过Ruby层的循环处理：

def self.add_flag_to_users(user_ids, flag)
  return if user_ids.empty?

  placeholders = user_ids.map { '?' }.join(',')
  UserFlag.connection.execute(
    <<~SQL,
      INSERT INTO user_flags (user_id, flag, created_at, updated_at)
      SELECT unnest(ARRAY[#{placeholders}]) AS user_id, ?, NOW(), NOW()
      WHERE NOT EXISTS (
        SELECT 1 FROM user_flags WHERE user_id = unnest(ARRAY[#{placeholders}]) AND flag = ?
      )
    SQL
    user_ids + [flag] + user_ids + [flag]
  )
end

注：确保user_flags表的user_id和flag字段有联合唯一索引，避免重复插入的同时加速查询。

额外优化建议 #

• 添加联合唯一索引：

# 在迁移文件中执行
add_index :user_flags, [:user_id, :flag], unique: true

• 增加异常捕获与日志：

class BatchProcessUsersJob < ApplicationJob
  queue_as :default

  def perform
    user_ids = [
      # 18,000 ids
    ]

    user_ids.each_slice(1000) do |batch_ids|
      begin
        UserFlag.add_flag_to_users(batch_ids, UserFlag::PRODUCT_TYPE_2)
        Rails.logger.info("Processed batch: #{batch_ids.size} users")
      rescue => e
        Rails.logger.error("Batch failed: #{e.message}")
        # 可选：重试失败批次
        BatchProcessUsersJob.perform_later(batch_ids)
      end
    end
  end
end

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