parallel-processing - Elixir Stream 中的 Task.async

Question

我想在一个大列表上做一个平行 map 。代码看起来有点像这样:

big_list
|> Stream.map(&Task.async(Module, :do_something, [&1]))
|> Stream.map(&Task.await(&1))
|> Enum.filter filter_fun

但我正在检查 Stream 实现，据我所知 Stream.map组合函数并将组合函数应用于流中的元素，这意味着序列是这样的:

取第一个元素

创建异步任务

等待它完成

取第二个元素...

在这种情况下，它不会并行执行。我是对的还是我错过了什么？

如果我是对的，那么这段代码呢？

Stream.map Task.async ...
|> Enum.map Task.await ...

这会并行运行吗？

Answer 1

第二个也没有做你想做的。您可以使用以下代码清楚地看到它:

defmodule Test do
  def test do
    [1,2,3]
    |> Stream.map(&Task.async(Test, :job, [&1]))
    |> Enum.map(&Task.await(&1))
  end

  def job(number) do
    :timer.sleep 1000
    IO.inspect(number)
  end
end

Test.test

您会看到一个数字，然后等待 1 秒，然后是另一个数字，依此类推。这里的关键是你想尽快创建任务，所以你不应该使用
懒惰 Stream.map根本。而是使用急切的 Enum.map在那时候:

|> Enum.map(&Task.async(Test, :job, [&1]))
|> Enum.map(&Task.await(&1))

另一方面，您可以使用 Stream.map等待的时候，只要你稍后做一些急切的操作，比如你的 filter .这样，等待将穿插在您可能对结果进行的任何处理中。