Rails+Sidekiq+SFTP环境下批量处理FTP文件的最优稳定方案

先直接给结论：第二种方案（仅将解析、存库、日志、删文件逻辑放入Sidekiq任务）的崩溃风险低得多，下面详细拆解原因和优化建议：

为什么不选第一种方案？ #

把整个Net::SFTP.start代码块塞进Sidekiq worker，相当于把一个长时间的批量任务丢给单个Sidekiq进程，这完全违背了Sidekiq「处理短生命周期任务」的设计初衷：

• SFTP长连接会被长时间占用，如果Sidekiq进程意外重启、被OOM杀死或者超时，SFTP服务器可能残留无效连接，甚至影响后续的连接请求。
• 所有文件处理都在一个worker里串行执行，只要中间某个文件处理失败（比如解析报错、数据库写入超时），整个批次的后续文件都会被中断，而且排查问题时很难定位到具体是哪个文件出了问题。
• 单个worker长时间占用资源（比如内存、数据库连接），会增加该进程崩溃的概率，而且Sidekiq的调度机制也无法高效处理这种长任务。

第二种方案的优势 #

让cron触发的Rake任务只做「轻量操作」：连接SFTP、列出文件、下载文件（到本地临时目录或者内存），然后把每个文件的处理逻辑拆成独立的Sidekiq任务并行执行，好处包括：

1. SFTP连接生命周期可控：Rake任务完成文件下载后就立刻关闭SFTP连接，不会出现长期占用连接的情况，避免连接泄漏和服务器资源浪费。
2. 故障隔离：每个文件的处理都是独立的Sidekiq任务，某个文件处理失败（比如解析错误、数据库写入失败），只会影响该任务，其他文件的处理不受干扰，而且Sidekiq还能自动重试失败的任务（你可以配置重试策略）。
3. 并行处理提升效率：20个文件不用串行等7分钟，Sidekiq可以同时启动多个worker并行处理，大幅缩短整体处理时间，减少系统长时间处于高负载的状态，自然降低崩溃风险。
4. 符合Sidekiq最佳实践：每个worker处理单个文件，任务生命周期短（20秒左右），Sidekiq的调度、监控和资源管理都能更高效地运作。

额外优化建议 #

为了进一步降低风险，你还可以做这些调整：

• 避免将大文件数据直接放入Sidekiq队列：如果文件较大，不要把文件内容作为参数传给worker，建议先把文件下载到本地临时目录，然后把临时文件路径作为参数传给worker（处理完后记得清理临时文件）。
• 添加文件处理锁机制：防止cron重复触发时，同一个文件被多个Rake任务重复下载处理。比如可以在SFTP服务器上把待处理文件先移动到「processing」目录再下载，或者用Redis给每个文件名加锁。
• 完善日志和监控：给每个Sidekiq任务添加独立的日志标记（比如文件名、任务ID），方便排查问题；同时监控Sidekiq的队列长度、任务失败率，及时发现异常。
• 合理设置Sidekiq并发数：根据你的数据库和服务器性能，调整Sidekiq的并发数，避免同时启动太多worker导致数据库连接耗尽或者SFTP服务器压力过大。

示例伪代码调整 #

调整后的Rake任务：

namespace :check_ftp do
  desc "Check every 30 minutes"
  task :fetch_orders => [:environment] do
    check_dir = '/dir'
    processing_dir = '/dir/processing'
    Net::SFTP.start(host, username, password: pass) do |sftp|
      # 先创建processing目录（如果不存在）
      sftp.mkdir!(processing_dir) unless sftp.dir.exist?(processing_dir)
      
      sftp.dir.entries(check_dir).each do |remote_file|
        next if remote_file.name == '.' || remote_file.name == '..' || remote_file.name == 'processing'
        
        remote_file_path = "#{check_dir}/#{remote_file.name}"
        processing_file_path = "#{processing_dir}/#{remote_file.name}"
        
        # 先移动文件到processing目录，防止重复处理
        sftp.rename!(remote_file_path, processing_file_path)
        
        # 下载到本地临时文件
        local_temp_path = "/tmp/sftp_orders/#{remote_file.name}"
        FileUtils.mkdir_p(File.dirname(local_temp_path))
        sftp.download!(processing_file_path, local_temp_path)
        
        # 将处理任务丢给Sidekiq
        OrderProcessingWorker.perform_async(local_temp_path, processing_file_path)
      end
    end
  end
end

Sidekiq Worker：

class OrderProcessingWorker
  include Sidekiq::Worker
  sidekiq_options retry: 3 # 失败后重试3次

  def perform(local_temp_path, remote_processing_path)
    begin
      # 读取文件内容
      file_data = File.read(local_temp_path)
      
      # 解析内容
      parsed_orders = parse_order_data(file_data)
      
      # 存入数据库
      Order.transaction do
        parsed_orders.each { |order| Order.create!(order) }
      end
      
      # 记录日志
      Rails.logger.info "Successfully processed order file: #{File.basename(local_temp_path)}"
      
      # 删除SFTP上的文件
      Net::SFTP.start(host, username, password: pass) do |sftp|
        sftp.remove!(remote_processing_path)
      end
    rescue => e
      Rails.logger.error "Failed to process file #{local_temp_path}: #{e.message}"
      raise # 让Sidekiq重试
    ensure
      # 清理本地临时文件
      File.delete(local_temp_path) if File.exist?(local_temp_path)
    end
  end

  private

  def parse_order_data(file_data)
    # 你的解析逻辑
  end
end

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