处理并发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 选项，允许您注入自定义线程池。然后，您可以使用高级抽象提供的异常处理。

如果您绝对坚持将作业直接发布到线程池而不是使用我们的高级抽象（我强烈反对），那么只需创建一个作业包装器即可。您可以在我们所有的高级抽象、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!