Ruby迭代处理文件时如何限制并发线程数规避资源不可用错误

• 第一种写法一次性为2.4万个文件创建独立线程，远超操作系统单进程可持有的线程数上限（通常单进程线程阈值在数百到数千区间），直接触发can't create thread, resource temporarily unavailable错误。同时代码中手动File.open打开文件后未主动关闭，会泄漏文件句柄，进一步加剧资源占用问题。
• 第二种写法每创建一个线程就立刻调用join阻塞主线程等待该线程执行完毕，执行逻辑和单线程串行完全一致，没有任何并发收益，因此处理速度没有提升。
• 最初设想的「查询系统最大线程数→按总文件数/最大线程数拆分批次→逐批执行」的方案存在明显缺陷：固定批次切分容易出现负载不均——如果某一批次存在几个处理耗时极长的文件，会导致同批次其他线程执行完任务后空等，浪费算力；同时系统最大线程数是进程可创建线程的理论上限，图片处理属于CPU+IO密集型任务，线程数开到理论上限会引发频繁的CPU上下文切换、磁盘IO争抢，反而会拖慢整体处理速度。

使用固定大小工作线程+任务队列的线程池模式实现，不需要手动拆分批次：初始化固定数量的工作线程，所有待处理任务放入线程安全的队列，工作线程执行完当前任务后自动从队列领取下一个任务，既不会超出系统资源阈值，也能保证所有线程全程处于工作状态，并发效率最高。

参数配置建议 #

• 不需要查询系统最大线程数，图片处理场景并发线程数设置为CPU核心数的1~2倍即可：机械硬盘环境设为等于CPU核心数，避免IO争抢；SSD环境可设为CPU核心数的2倍。可以通过标准库Etc.nprocessors直接获取当前机器的CPU核心数，无需硬编码。
• 进度统计不要每次调用Dir.children("transparent").count遍历目录，该操作每执行一次就要扫描整个输出目录，文件量越大额外IO开销越高，直接用内存计数器统计即可。

可直接运行的实现代码 #

require 'image_processing/mini_magick'
require 'fileutils'
require 'etc'
require 'thread'

# 初始化配置
INPUT_DIR = "original"
OUTPUT_DIR = "transparent"
Dir.mkdir(OUTPUT_DIR) unless Dir.exist?(OUTPUT_DIR)
# 读取所有待处理图片文件
frames = Dir.children(INPUT_DIR).select { |f| File.file?(File.join(INPUT_DIR, f)) }
total = frames.count
# 配置并发线程数：SSD可调整为 Etc.nprocessors * 2
MAX_WORKERS = Etc.nprocessors

# 线程安全的输出锁与进度计数器，避免多线程输出错乱、计数冲突
mutex = Mutex.new
processed = 0

# 初始化任务队列，塞入所有待处理任务
task_queue = Queue.new
frames.each { |img| task_queue << img }

# 启动固定数量的工作线程
workers = MAX_WORKERS.times.map do
  Thread.new do
    # 循环从队列取任务，直到队列清空
    while img = task_queue.pop(true) rescue nil
      begin
        # 直接传文件路径，由MiniMagick自行管理文件句柄，避免泄漏
        pipeline = ImageProcessing::MiniMagick
          .source(File.join(INPUT_DIR, img))
          .append("-fuzz", "30%")
          .append("-transparent", "#ff00fe")
        result = pipeline.call

        # 提取帧号保存结果
        frame_num = img.split("_")[2]
        output_path = File.join(OUTPUT_DIR, "image_transparent_#{frame_num}")
        FileUtils.cp(result.path, output_path)

        # 线程安全打印进度
        mutex.synchronize do
          processed += 1
          puts "Finish processing: #{img}"
          puts "Progress: #{processed} / #{total}"
          puts "---"
        end
      rescue => e
        mutex.synchronize { puts "Process #{img} failed: #{e.message}" }
      ensure
        # 确保临时文件被清理，避免占用磁盘空间
        result&.close! if defined?(result)
      end
    end
  end
end

# 等待所有工作线程执行完毕
workers.each(&:join)
puts "All #{total} images processed completed."

额外优化点 #

• 加入了异常捕获逻辑，单个图片损坏、格式异常不会导致整个脚本崩溃，错误信息打印后会继续处理剩余文件。
• 新增临时文件清理逻辑，MiniMagick处理生成的临时文件会在任务完成后自动删除，长时间运行不会占满系统临时目录。
• 所有多线程共享的输出、计数操作都加了互斥锁，不会出现控制台输出乱码、进度计数不准的问题。
• 不需要引入任何第三方依赖，所有用到的库都是Ruby标准库+已在使用的ImageProcessing、MiniMagick，可直接运行。

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