产生多个并行任务答案

【问题标题】：spawning multiple parallel tasks产生多个并行任务
【发布时间】：2012-03-14 06:54:27
【问题描述】：

我有 M 个要处理的任务和 N 个可用的并行处理资源（想想 Herko 或 EC2 实例上的工作线程），其中 M >> N。

我可以推出自己的系统，但似乎已经有一个经过调试的包或 gem：你有什么建议？（现在我想起来了，我可以折磨 Delayed::Job 来这样做。）

这些任务几乎可以用任何语言编写——即使是 shell 脚本也可以完成这项工作。 “母船”是带有 PostgreSQL 数据库的 Ruby On Rails。基本思想是，当资源准备好处理任务时，它会向母船询问队列中下一个未处理的任务并开始处理它。如果作业失败，则在放弃之前重试几次。结果可以进入平面文件或写入 PostgreSQL 数据库。

（而且，不，这不是为了生成垃圾邮件。我正在研究几个大型社交网络的degree distribution。）

【问题讨论】：

标签： ruby ruby-on-rails-3 parallel-processing

【解决方案1】：

正如你所说，我认为这是一个延迟工作 https://github.com/collectiveidea/delayed_job 或 resque https://github.com/defunkt/resque 工作。

【讨论】：

同意。我不明白为什么这个解决方案会被认为是“折磨”。
我的意思是，除了 D::J 或 Resque 提供的工作处理之外，还有很多细节需要注意，例如分配任务并处理错误。但这很可能是我要走的路。
接受这个答案。不过，我很有可能会使用 Amazon 的 SQS 作为整个工作流程的一部分。

【解决方案2】：

这将是你自己的，但如果你的并行任务不是资源密集型的，它是一个相当快速的解决方案。另一方面，如果它们是资源密集型的，您将希望实现更强大的东西。

您可以使用Process::fork（如果进程在 ruby 中）、Process::exec 或Process::spawn（如果进程在其他东西中）启动每个线程。然后使用Process::waitall 完成子流程。

下面，我使用Hash 来保存函数本身以及 PID。这绝对可以改进。

# define the sub-processes
sleep_2_fail = lambda { sleep 2; exit -1; }
sleep_2_pass = lambda { sleep 2; exit 0; }
sleep_1_pass = lambda { sleep 1; exit 0; }
sleep_3_fail = lambda { sleep 3; exit -1; }

# use a hash to store the lambda's and their PID's
sub_processes = Hash.new

# add the sub_processes to the hash
#  key = PID
#  value = lambda (can use to be re-called later on)
sub_processes.merge! ({ Process::fork { sleep_2_fail.call } => sleep_2_fail })
sub_processes.merge! ({ Process::fork { sleep_2_pass.call } => sleep_2_pass })
sub_processes.merge! ({ Process::fork { sleep_1_pass.call } => sleep_1_pass })
sub_processes.merge! ({ Process::fork { sleep_3_fail.call } => sleep_3_fail })

# starting time of the loop
start = Time.now

# use a while loop to wait at most 10 seconds or until
# the results are empty (no sub-processes)
while ((results = Process.waitall).count > 0 && Time.now - start < 10) do
  results.each do |pid, status|
    if status != 0
       # again add the { PID => lambda } to the hash
       sub_processes.merge! ( { Process::fork { sub_processes[pid].call } => sub_processes[pid] } )
    end
    # delete the original entry
    sub_processes.delete pid
  end
end

waitall 上的 ruby-doc 很有帮助。

【讨论】：

【解决方案3】：

听起来你想要一个工作处理器。看看Gearman http://gearman.org/ 与语言完全无关。

这里是红宝石信息http://gearmanhq.com/help/tutorials/ruby/getting_started/

【讨论】：