Ruby Thread Pooling - 我做错了什么？(Ruby Thread Pooling - What am I doing wrong?)

 我在Postgres数据库的Content表中有250万条记录，我需要遍历这250万条记录中的每条记录并执行一些操作（其中很多都是慢的）并根据内容更新记录。我一直聚集在一起 - 一切都行得通 - 这个问题，需要血腥的运行。 
 
 我遇到了几篇关于多线程这类工作的文章（我之前在C中做过这个，但从来没有使用Ruby）以及在Ruby中使用线程的优点和缺点，尽管有这些缺点，我可以得到的2000个线程显着发生比没有线程运行更快，但我一次只能关闭2000，限制我实际上能够更新所有250万条记录。 这是我的代码： 
 
Content.all.each do |content|
  threads << Thread.new do
    grab_and_store(content)
  end
  index += 1
  index % 100 == 0 ? (puts index) : nil
end
threads.map(&:join)
 
 我还读到了关于线程池的问题，一旦完成了原来的工作，就使用相同的线程来完成其他工作，但我似乎无法让它工作。 这是我的代码： 
 
POOL_SIZE = 1000

jobs = Queue.new
Content.all.each{ |x| jobs.push x }

workers = (POOL_SIZE).times.map do
  Thread.new do
    begin
      while x = jobs.pop(true)
        grab_and_store(x)
      end
    rescue ThreadError
    end
  end
end
workers.map(&:join)
 
 当我运行这个时，我得到一个错误，我无法在nil类上执行.join，这意味着工作者在此结束时是零。 但是，当我采用我基于此的代码（如下所示和源代码 ）并运行它完美无缺。 我似乎无法弄清楚我的破坏在哪里/如何最好地实现线程池以阻止我的代码在2000线程之后耗尽资源。 
 
 谢谢！ 
 
 PS继承我使用的教程中的代码： 
 
require 'thread'
work_q = Queue.new
(0..50).to_a.each{|x| work_q.push x }
workers = (0...4).map do
  Thread.new do
    begin
      while x = work_q.pop(true)
        50.times{print [128000+x].pack "U*"}
      end
    rescue ThreadError
    end
  end
end; "ok"
workers.map(&:join); "ok"
 
 更新： 
 
 根据安东尼的回答，我发现自己使用了以下大块代码，使用他推荐的ruby-thread gem，它很快就会运行给定的内容（样本大小为1000），但是当我检查控制台时，它似乎只保存了大约20最大 这是代码： 
 
pool = Thread.pool(5)

@ids = []
arr = Content.where(needs_update: true)[0...1000]

puts "Starting With Sample 1000"

arr.each do |content|
  pool.process do
    grab_and_store(content)
  end
  index += 1
  index % 100 == 0 ? (puts index) : nil
end

pool.shutdown

I have 2.5 million records in the Content table of my Postgres Database, I need to go through each of those 2.5 million records and perform a number of actions (many of which are slow by themselves) and update the record at the end based on what I have gathered along the way - that all works - the problem, it takes for bloody ever to run.
 
I came across a couple of articles talking about multithreading such jobs (I have done this before in C, but never Ruby) and the pros and cons of using threads in Ruby, still despite those cons, the 2000 threads I can get off happen significantly faster than running without threading, but I can only get 2000 off at one time, limiting me from actually being able to update all 2.5 million records. Here is the code I had for that:
 
Content.all.each do |content|
  threads << Thread.new do
    grab_and_store(content)
  end
  index += 1
  index % 100 == 0 ? (puts index) : nil
end
threads.map(&:join)
 
I also read about thread pooling, using the same threads to do other jobs once they have completed their original one, but I can't seem to get it to work. Here is the code that I had:
 
POOL_SIZE = 1000

jobs = Queue.new
Content.all.each{ |x| jobs.push x }

workers = (POOL_SIZE).times.map do
  Thread.new do
    begin
      while x = jobs.pop(true)
        grab_and_store(x)
      end
    rescue ThreadError
    end
  end
end
workers.map(&:join)
 
When I run this I get an error that I can't execute .join on a nil class, which would mean that workers is nil at the end of this. But when I take the code that I based this off of (shown below, and source) and run that it works perfectly. I can't seem to figure out where mine is breaking / how to best implement the thread pool to stop my code from running out of resources after 2000 threads.
 
Thanks!
 
P.S. Heres the code from the tutorial I used:
 
require 'thread'
work_q = Queue.new
(0..50).to_a.each{|x| work_q.push x }
workers = (0...4).map do
  Thread.new do
    begin
      while x = work_q.pop(true)
        50.times{print [128000+x].pack "U*"}
      end
    rescue ThreadError
    end
  end
end; "ok"
workers.map(&:join); "ok"
 
Update:
 
Per Anthony's answer I found myself with the following chunk of code, using the ruby-thread gem he recommended, it runs through the given Content really quickly (it's a sample size of 1000), but when I check console it appears to have only saved around 20 max. Here's the code:
 
pool = Thread.pool(5)

@ids = []
arr = Content.where(needs_update: true)[0...1000]

puts "Starting With Sample 1000"

arr.each do |content|
  pool.process do
    grab_and_store(content)
  end
  index += 1
  index % 100 == 0 ? (puts index) : nil
end

pool.shutdown

原文：https://stackoverflow.com/questions/32593289