Elixir 进程限制？

Question

我想创建一个类似于这样的 Elixir 代码:

def infinite_loop(created_workers \\ []) do
  case next_from_queue do
    {:ok, queue_msg} ->
      new_worker = Task.async(fn -> crawling(queue_msg) end)
      infinite_loop([new_worker | created_workers])
    {:error, :empty} ->
      created_workers.map(&Task.await/1)
  end
end

假如说:

crawling函数将创建另一个 3 Task

每crawling worker 可以花 3 秒运行

queue可能有数百万条消息

我怎么知道 Elixir 上并行进程的限制是多少？我怎样才能让它不破裂？

Answer 1

我建议使用 Task.async_stream为了这。 Task.async_stream允许您并行处理流，同时限制并行运行的任务数量。虽然 Erlang 20 中进程数的默认限制是 262144，但如果您正在抓取网站，您可能需要一个更低的限制。

您可以使用 Stream.iterate 从不断返回新项目的函数中创建流。 :

stream =
  Stream.iterate(next_from_queue(), fn _ -> next_from_queue() end)
  |> Stream.take_while(fn {:ok, _} -> true; {:error, :empty} -> false end)

既然你想停在 {:error, :empty} ，我们使用 Stream.take_while停止流。

然后使用 Task.async_stream像这样:

stream
|> Task.async_stream(fn {:ok, queue_msg} ->
  crawling(queue_msg)
end, max_concurrency: 16)

这将并行运行最多 16 个任务的流。最终结果将是 crawling(queue_msg) 的所有返回值的列表.