Ruby 3 从多个预定纤程中收集结果

发布于12月08日

Ruby3引入了Fiber.schedule来并发分派异步任务.

类似于this question中提出的问题(这是关于线程并发的)，我想要一种在纤程调度器上启动多个并发任务的方法，一旦它们都被调度好，就等待它们的组合结果，这在某种程度上相当于JavaScript中的Promise.all.

我可以想出这个天真的方法:

require 'async'

def io_work(t)
  sleep t
  :ok
end

Async do
  results = []

  [0.1, 0.3, 'cow'].each_with_index do |t, i|
    n = i + 1
    Fiber.schedule do
      puts "Starting fiber #{n}\n"
      result = io_work t
      puts "Done working for #{t} seconds in fiber #{n}"
      results << [n, result]
    rescue
      puts "Execution failed in fiber #{n}"
      results << [n, :error]
    end
  end

  # await combined results
  sleep 0.1 until results.size >= 3

  puts "Results: #{results}"
end

有没有一种更简单的 struct 可以起到同样的作用？

推荐答案

由于已经安排了Async个任务，我不确定您是否需要所有这些任务.

如果你只是想等所有的项目都完成，你可以使用Async::Barrier

示例:

require 'async'
require 'async/barrier'


def io_work(t)
  sleep t
  :ok
end

Async do
  barrier = Async::Barrier.new
  results = []
  [1, 0.3, 'cow'].each.with_index(1) do |data, idx|
    barrier.async do 
      results << begin
        puts "Starting task #{idx}\n"
        result = io_work data
        puts "Done working for #{data} seconds in task #{idx}"
        [idx,result]
      rescue
        puts "Execution failed in task #{idx}"
        [idx, :error]
      end          
    end 
  end
  barrier.wait
  puts "Results: #{results}"
end

基于此将输出的sleep个值

Starting task 1
Starting task 2
Starting task 3
Execution failed in task 3
Done working for 0.3 seconds in task 2
Done working for 1 seconds in task 1
Results: [[3, :error], [2, :ok], [1, :ok]]

barrier.wait将等待所有异步任务完成，如果没有它，输出将如下所示

Starting fiber 1
Starting fiber 2
Starting fiber 3
Execution failed in fiber 3
Results: [[3, :error]]
Done working for 0.3 seconds in fiber 2
Done working for 1 seconds in fiber 1