Ruby多进程下载同一文件的竞态条件处理方案咨询

一、修复文件锁的竞态问题 #

你之前用文件锁出现竞态，大概率是实现方式不对。正确的做法是用排他锁（独占锁），并且拿到锁后必须二次校验文件状态，因为等锁的过程中可能已有进程完成下载。

Ruby里可以这么实现：

def ensure_file_ready(file_path, gs_url)
  lock_file = "#{file_path}.lock"
  # 阻塞获取排他锁，直到拿到为止
  File.open(lock_file, 'w') do |f|
    f.flock(File::LOCK_EX)
    # 拿到锁后再次检查文件是否存在且完整
    unless File.exist?(file_path) && file_valid?(file_path)
      # 从Google Storage下载到目标路径
      download_from_gs(gs_url, file_path)
    end
  end
  # 文件关闭时自动释放锁，哪怕下载出错也不会死锁
end

# 辅助方法：校验文件完整性（比如对比Google Storage上的文件大小/哈希）
def file_valid?(file_path)
  gs_file_size = get_gs_file_size(file_path)
  File.size(file_path) == gs_file_size
end

二、原子文件写入（从根源避免半完成文件） #

下载时先写入临时文件，完成后再原子重命名到目标路径——Unix/Linux系统里File.rename是原子操作，不会出现“文件存在但不完整”的情况。

示例逻辑：

def ensure_file_ready(file_path, gs_url)
  temp_file = "#{file_path}.tmp"
  # 先检查目标文件是否可用
  return if File.exist?(file_path) && file_valid?(file_path)

  # 锁定临时文件，避免多进程同时写
  File.open(temp_file, 'w') do |f|
    f.flock(File::LOCK_EX)
    # 二次校验，防止等锁期间其他进程已经完成下载
    return if File.exist?(file_path) && file_valid?(file_path)
    
    # 下载到临时文件
    download_from_gs(gs_url, temp_file)
    # 校验临时文件完整性
    if file_valid?(temp_file)
      # 原子替换目标文件
      File.rename(temp_file, file_path)
    else
      # 下载失败，清理临时文件
      File.unlink(temp_file)
      raise "下载的文件不完整"
    end
  end
rescue Errno::EEXIST
  # 临时文件被其他进程锁定，短延时后重试
  sleep 0.1
  retry
end

三、任务队列方案（完全可行） #

把下载逻辑剥离到任务队列（比如Sidekiq、Resque），利用队列的唯一性保证同一文件只被下载一次，彻底避开竞态。

具体步骤：

1. 请求时先检查文件状态，可用则直接使用；
2. 不可用则发送下载任务到队列，然后等待任务完成（或返回重试响应让客户端稍后再试）；
3. 队列worker通过唯一任务ID避免重复执行。

伪代码示例：

# 路由逻辑
get '/process-file' do
  file_path = '/path/to/target.file'
  gs_url = 'gs://bucket/file'

  if File.exist?(file_path) && file_valid?(file_path)
    process_file(file_path)
  else
    # 发送唯一下载任务
    job_id = DownloadFileJob.perform_async(gs_url, file_path)
    # 等待任务完成（或返回202让客户端轮询）
    wait_for_job_completion(job_id)
    process_file(file_path)
  end
end

# Sidekiq任务
class DownloadFileJob
  include Sidekiq::Job
  # 开启任务唯一性，5分钟内同一任务只执行一次
  sidekiq_options unique: true, unique_expiration: 300

  def perform(gs_url, file_path)
    temp_file = "#{file_path}.tmp"
    download_from_gs(gs_url, temp_file)
    if file_valid?(temp_file)
      File.rename(temp_file, file_path)
    else
      File.unlink(temp_file)
      raise "文件下载不完整"
    end
  end
end

这个方案的好处是不占用请求线程，但需要额外部署队列服务，增加了一点架构复杂度。

四、预加载文件到Puma Workers #

如果文件不经常更新，可以在Puma启动时，让master进程提前下载好文件，fork worker时文件会被继承到所有worker里，彻底避免运行时竞态。

修改config/puma.rb：

# master进程fork worker前执行，只下载一次
before_fork do
  file_path = '/path/to/target.file'
  gs_url = 'gs://bucket/file'
  unless File.exist?(file_path) && file_valid?(file_path)
    download_from_gs(gs_url, file_path)
  end
end

适合静态文件场景，优点是简单高效，缺点是文件更新时需要重启Puma。

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