ruby - 处理并发 ruby​​ 线程池中的异常

Question

如何处理并发 ruby​​ 线程池中的异常 (http://ruby-concurrency.github.io/concurrent-ruby/file.thread_pools.html)？

例子:

pool = Concurrent::FixedThreadPool.new(5) 
pool.post do
  raise 'something goes wrong'
end

# how to rescue this exception here

---

更新:

这是我的代码的简化版本:

def process
  pool = Concurrent::FixedThreadPool.new(5)

  products.each do |product|
    new_product = generate_new_product

    pool.post do
      store_in_db(new_product) # here exception is raised, e.g. connection to db failed
    end
  end

  pool.shutdown
  pool.wait_for_terminaton
end

所以我想要实现的是在出现任何异常时停止处理(中断循环)。

这个异常也在更高级别的应用程序中被拯救并且执行了一些清理工作(比如将模型状态设置为失败并发送一些通知)。

Answer 1

以下答案来自jdantonio 来自这里https://github.com/ruby-concurrency/concurrent-ruby/issues/616

"大多数应用程序不应该直接使用线程池。线程池是一种低级抽象，供内部使用。该库中的所有高级抽象(Promise、Actor 等)都将作业发布到全局线程池，并且都提供异常处理。只需选择最适合您的用例的抽象并使用它。

如果您觉得需要配置自己的线程池而不是使用全局线程池，您仍然可以使用高级抽象。它们都支持 :executor 选项，允许您注入(inject)自定义线程池。然后，您可以使用高级抽象提供的异常处理。

如果您绝对坚持将作业直接发布到线程池而不是使用我们的高级抽象(我强烈反对)，那么只需创建一个作业包装器。您可以在我们所有的高级抽象、Rails ActiveJob、Sucker Punch 和其他使用我们线程池的库中找到作业包装器的示例。”

那么使用 Promises 的实现怎么样？ http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/Promise.html在您的情况下，它看起来像这样:

promises = []
products.each do |product|
  new_product = generate_new_prodcut

  promises << Concurrent::Promise.execute do 
    store_in_db(new_product)
  end
end

# .value will wait for the Thread to finish.
# The ! means, that all exceptions will be propagated to the main thread
# .zip will make one Promise which contains all other promises.
Concurrent::Promise.zip(*promises).value!