Ruby中大量任务场景下CSV文件更新优化及原子写入实现问询

嘿，这个场景我太熟悉了——用CSV当任务队列处理十万级任务，每次全量写入确实能把硬盘折腾得够呛，而且原脚本还有递归栈溢出的隐患，咱们一步步来优化：

首先得先指出原脚本里的几个坑：

• 递归调用resolve方法，十万次递归绝对会触发栈溢出，Ruby默认栈大小撑不住这么多次
• 每处理一行就把整个CSV重新写一遍，IO开销拉满，大文件反复写入不仅耗硬盘，还慢
• 没有错误处理，任务出错直接中断，之前的进度虽然写了，但错误排查也麻烦

优化方案1：原子写入+内存缓存，既安全又高效 #

核心思路是：把CSV一次性加载到内存，处理一行就更新内存数据，然后用原子写入的方式替换原文件——这样既避免了中途崩溃导致文件损坏，也减少了重复加载CSV的开销。

修改后的代码如下：

require 'csv'
require 'json'
require 'fileutils'

class CSVResolver
  attr_accessor :csv_path

  def initialize(csv_path:)
    self.csv_path = csv_path
    # 只加载一次CSV到内存，避免反复读取大文件
    @csv_data = CSV.read(csv_path)
  end

  def resolve
    @csv_data.each_with_index do |row, index|
      # 跳过已经完成的任务
      next if row[1].present?

      begin
        # 执行你的耗时任务（这里模拟一下）
        json = very_expensive_task_and_error_prone(row[0])
        # 更新当前行的结果路径
        row[1] = "/data/#{index}.json"
        # 写入结果JSON文件
        File.write(row[1], JSON.pretty_generate(json))
        
        # 原子写入更新后的CSV，保证数据安全
        atomic_write_csv
        puts "任务##{index+1}完成，进度已保存"
      rescue StandardError => e
        # 捕获错误，避免整个流程中断
        puts "任务##{index+1}失败：#{e.message}"
        next
      end
    end
  end

  private

  def atomic_write_csv
    # 创建临时文件，和原文件同目录才能保证原子替换
    temp_file_path = "#{csv_path}.tmp"
    # 把内存中的CSV数据写入临时文件
    CSV.open(temp_file_path, "wb") do |csv|
      @csv_data.each { |row| csv << row }
    end
    # 系统级原子替换，要么完全替换成功，要么原文件保持不变
    FileUtils.mv(temp_file_path, csv_path, force: true)
  end

  def very_expensive_task_and_error_prone(task_content)
    # 这里替换成你的实际任务逻辑
    { task: task_content, processed_at: Time.now.iso8601 }
  end
end

这个方案的好处：

• 原子写入：用临时文件+FileUtils.mv（系统提供的原子操作），就算脚本中途崩溃，原CSV文件也不会损坏，进度只会停在上一次成功写入的位置
• 减少IO：只在初始化时读一次CSV，后续都是内存操作，写入时虽然还是全量，但比原脚本每次都重新读再写要高效得多
• 错误容错：添加了异常捕获，单个任务失败不会影响整个队列的处理
• 避免栈溢出：把递归改成了循环遍历，十万次循环完全没问题

优化方案2：拆分进度记录，彻底告别大文件IO #

如果十万级任务全量写入还是觉得IO压力大，可以换个思路：原CSV只存待办任务，单独用一个小文件记录已完成的进度。这样每次只需要写入很小的进度文件，IO开销几乎可以忽略。

示例代码大概是这样：

require 'csv'
require 'json'
require 'fileutils'

class CSVResolver
  attr_accessor :csv_path, :progress_path

  def initialize(csv_path:, progress_path: "task_progress.json")
    self.csv_path = csv_path
    self.progress_path = progress_path
    # 加载已完成的任务进度
    @completed_tasks = load_progress
    # 加载待办任务列表
    @tasks = CSV.read(csv_path)
  end

  def resolve
    @tasks.each_with_index do |row, index|
      task_id = index + 1
      # 跳过已完成的任务
      next if @completed_tasks.key?(task_id.to_s)

      begin
        json = very_expensive_task_and_error_prone(row[0])
        output_file = "/data/#{task_id}.json"
        File.write(output_file, JSON.pretty_generate(json))
        
        # 更新进度并写入
        @completed_tasks[task_id.to_s] = output_file
        save_progress
        puts "任务##{task_id}完成，进度已保存"
      rescue StandardError => e
        puts "任务##{task_id}失败：#{e.message}"
        next
      end
    end
  end

  private

  def load_progress
    # 加载已完成的任务进度，没有就返回空哈希
    return {} unless File.exist?(progress_path)
    JSON.parse(File.read(progress_path))
  end

  def save_progress
    # 同样用原子写入保证进度文件的安全
    temp_path = "#{progress_path}.tmp"
    File.write(temp_path, JSON.pretty_generate(@completed_tasks))
    FileUtils.mv(temp_path, progress_path, force: true)
  end

  def very_expensive_task_and_error_prone(task_content)
    { task: task_content, processed_at: Time.now.iso8601 }
  end
end

这个方案的优势更明显：

• 进度文件极小，每次写入的IO成本几乎可以忽略
• 原CSV文件完全不需要修改，避免了大文件的读写操作
• 就算任务中断，重启脚本时只需要加载进度文件就能跳过已完成的任务，完全不影响

额外的小技巧 #

• 批量写入进度：如果任务成功率很高，可以每完成100个任务再写入一次进度，进一步减少IO（但要权衡，比如崩溃会丢失最近100个任务的进度）
• 换用更快的CSV库：比如fastcsv，比Ruby标准库的CSV读写速度快不少，适合处理超大文件
• 并行处理：如果任务是IO密集型（比如调用第三方API），可以用多线程或进程并行处理，但要注意进度文件的写入需要加锁，避免并发写入冲突

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