Ruby 中的线程安全枚举器

Question

TLDR：Ruby 中是否有 Enumerator 类的线程安全版本？

---

我想要做什么：

我想在 Ruby On Rails 应用程序中同时运行一个方法。该方法应该创建一个包含来自站点的报告的 zip 文件，其中 zip 中的每个文件都是 PDF。从html到PDF的转换有点慢，所以想要多线程。

我希望如何做到这一点：

我想使用 5 个线程，所以我想我会在线程之间有一个共享的枚举器。每个线程都会从 Enumerator 中弹出一个值，并对它运行 do stuff。这是我认为它会起作用的方式：

t = Zip::OutputStream::write_buffer do |z|
  mutex = Mutex.new
  gen = Enumerator.new{ |g|
    Report.all.includes("employee" => ["boss", "client"], "projects" => {"project_owner" => ["project_team"]}).find_each do |report|
      g.yield report
    end
  }
  5.times.map {
    Thread.new do
      begin
        loop do
          mutex.synchronize  do
            @report = gen.next
          end
          title = @report.title + "_" + @report.id.to_s
          title += ".pdf" unless title.end_with?(".pdf")
          pdf = PDFKit.new(render_to_string(:template => partial_url, locals: {array: [@report]},
                                            :layout => false)).to_pdf
          mutex.synchronize  do
            z.put_next_entry(title)
            z.write(pdf)
          end
        end
      rescue StopIteration
        # do nothing
      end
    end
  }.each {|thread| thread.join }
end

我尝试的时候发生了什么：

当我运行上面的代码时，我得到了以下错误：

FiberError at /generate_report
fiber called across threads

经过一番搜索，我遇到了this post，它建议我使用队列而不是枚举器，因为队列是线程安全的，而枚举器不是。虽然这对于非 Rails 应用程序可能是合理的，但对我来说这是不切实际的。

为什么我不能只使用队列：

Rails 4 ActiveRecord 的好处是它不会加载查询，直到它们被迭代。而且，如果您使用像find_each 这样的方法对其进行迭代，它会以 1000 个为单位进行迭代，因此您不必一次将整个表存储在 ram 中。我正在使用的查询结果：Report.all.includes("employee" => ["boss", "client"], "projects" => {"project_owner" => ["project_team"]}) 很大。很大。而且我需要能够即时加载它，而不是执行以下操作：

gen = Report.all.includes("employee" => ["boss", "client"], "projects" => {"project_owner" => ["project_team"]}).map(&queue.method(:push))

这会将整个查询加载到内存中。

最后的问题：

是否有线程安全的方式来做到这一点：

gen = Enumerator.new{ |g|
        Report.all.includes(...).find_each do |report|
          g.yield report
        end
}

这样我就可以跨多个线程从gen 弹出数据，而不必将整个Report（以及所有包含）表加载到内存中？

Answer 1

如果您在填满队列之前启动工作线程，它们将在您填满队列时开始消耗队列，并且因为根据经验 - 网络比 CPU 慢，每个批次应该（大部分）由下一批到达的时间：

queue = Queue.new

t1 = Thread.new do
  while !queue.empty?
    p queue.pop(true)
    sleep(0.1)
  end
end
t2 = Thread.new do
  while !queue.empty?
    p queue.pop(true)
    sleep(0.1)
  end
end

(0..1000).map(&queue.method(:push))

t1.join
t2.join

如果这仍然太慢，您可以选择使用SizedQueue，如果队列达到足够大的大小，它将阻止push：

queue = SizedQueue.new(100)

t1 = Thread.new do
  while !queue.empty?
    p "#{queue.pop(true)} - #{queue.size}"
    sleep(0.1)
  end
end
t2 = Thread.new do
  while !queue.empty?
    p queue.pop(true)
    sleep(0.1)
  end
end
(0..300).map(&queue.method(:push))
t1.join
t2.join